数据同步的艺术:探索PostgreSQL和Redis的一致性策略

2024-09-01 16:52

本文主要是介绍数据同步的艺术:探索PostgreSQL和Redis的一致性策略,希望对大家解决编程问题提供一定的参考价值,需要的开发者们随着小编来一起学习吧!

作者:后端小肥肠

🍇 我写过的文章中的相关代码放到了gitee,地址:xfc-fdw-cloud: 公共解决方案

🍊 有疑问可私信或评论区联系我。

🥑  创作不易未经允许严禁转载。

1. 前言

在当今高度数字化的世界中,应用程序需要处理海量数据并提供快速响应。为了应对这一挑战,使用Redis作为缓存来减少对数据库的直接访问已经成为一种广泛采用的策略。这种方法不仅能够显著提升应用性能,还能有效降低数据库负载。然而,当我们将PostgreSQL作为主数据库,Redis作为缓存层时,一个关键问题随之而来:如何确保这两个系统之间的数据保持一致?

本文将深入探讨PostgreSQL和Redis之间的数据同步策略,旨在帮助开发者和架构师构建既高效又可靠的数据架构。本文将:

  1. 分析导致数据不一致的根本原因
  2. 探讨各种同步策略的优缺点
  3. 提供实际可行的解决方案

通过本文,读者将获得设计和实现健壮的数据同步机制所需的相关知识,从而在高并发环境中构建更可靠的应用系统。

2. 数据为什么会不一致?

在现代高并发应用中,确保数据的一致性是一个复杂而关键的挑战。让我们深入探讨导致数据不一致的主要原因,以及为什么在PostgreSQL和Redis之间同步数据如此重要。

2.1 数据库:并发访问的瓶颈

在高并发的业务场景下,数据库往往成为系统中最薄弱的环节:

  • 连接限制数据库能够同时处理的连接数是有限的,高并发下很容易达到这个限制。
  • 锁竞争:并发写操作可能导致严重的锁竞争,大幅降低系统吞吐量。
  • 资源消耗每个数据库操作都消耗CPU和I/O资源,并发请求增加会导致资源迅速耗尽。
  • 查询性能复杂查询在高并发下可能导致数据库响应时间显著增加。

这些因素综合导致数据库在高并发场景下性能急剧下降,成为整个系统的瓶颈。

2.2 Redis:缓解数据库压力的关键

为了应对数据库的局限性,引入Redis作为缓存层成为一种普遍策略:

  • 减轻数据库负载将热点数据存储在Redis中,大幅减少对数据库的直接访问。
  • 提高响应速度Redis的内存存储特性使得数据访问速度远快于传统数据库。
  • 支持高并发Redis的单线程模型和非阻塞I/O使其能够高效处理大量并发请求。
  • 数据结构多样性Redis提供了多种数据结构,适用于不同的缓存场景。

使用Redis作为缓冲层,让请求首先访问Redis而不是直接访问PostgreSQL等数据库,可以显著提升系统的整体性能和并发处理能力。

2.3 数据一致性挑战

虽然引入Redis缓存解决了许多性能问题,但也带来了新的挑战,特别是在数据更新方面:

  1. 读取操作从Redis缓存读取数据通常不会引发严重的一致性问题,因为它只涉及单一数据源。
  2. 写入操作当需要更新数据时,问题变得复杂。我们需要同时更新数据库(PostgreSQL)和缓存(Redis),这个过程中很容易出现数据不一致:
    • 更新时序数据库和缓存的更新顺序会影响数据一致性。
    • 部分失败如果更新过程中发生错误,可能导致数据库和缓存状态不一致。
    • 并发更新多个客户端同时更新同一数据可能导致意外的结果。
  3. 缓存失效决定何时使缓存中的数据失效也是一个挑战,过早失效会降低缓存效果,过晚失效则可能导致数据不一致。

2.4 典型的数据读取流程

为了更好地理解这个问题,让我们看一下典型的数据读取流程:

  1. 应用程序接收到数据读取请求。
  2. 首先检查Redis缓存中是否存在所需数据。
  3. 如果Redis中存在数据(缓存命中),直接返回缓存数据。
  4. 如果Redis中不存在数据(缓存未命中),则从PostgreSQL数据库读取数据。
  5. 将从数据库读取的数据写入Redis缓存,以便后续访问。
  6. 返回数据给应用程序。

这个流程在处理读取操作时通常运作良好,但一旦涉及数据更新,就需要额外的机制来确保PostgreSQL和Redis之间的数据一致性。

在接下来的章节中,我们将深入探讨各种同步策略,以及如何在保证数据一致性的同时,维持系统的高性能和可扩展性。

3. redis与Postgres数据同步策略介绍

3.1 先删除缓存后更新数据库

在这种策略中,系统首先删除Redis中的缓存,然后更新数据库。这种方法可能引起如下问题:

  1. 读取过程中的数据不一致如果在缓存被删除后、数据库更新之前,另一个线程尝试读取数据,它会发现缓存为空。此时,它可能会从数据库中读取旧数据并将其写回缓存,导致所谓的脏缓存
  2. 数据更新后的不一致性当数据库最终更新后,缓存中的数据已经是旧数据,从而导致缓存与数据库之间的不一致。

3.2 先更新数据库后删除缓存

这种策略先对数据库进行更新,然后再删除缓存。这样做的潜在问题包括:

  1. 事务失败导致的缓存未删除如果在数据库更新后、删除缓存之前,执行更新的线程出现故障,这会导致缓存未被及时删除。
  2. 并发导致的数据不一致由于数据库的写入和缓存的删除是两个独立操作,无法保证它们之间的执行顺序,可能会导致读取操作在缓存删除前读到旧缓存,进而出现数据不一致。

3.3 延时双删策略

为了解决上述提到的同步问题,可以采用延时双删策略:

  1. 删除缓存首先删除Redis中的相关数据缓存。
  2. 更新数据库然后更新数据库,确保数据的准确性。
  3. 休眠延迟执行短暂的延时(例如500毫秒),确保所有相关数据库操作都已完成。
  4. 再次删除缓存最后再次删除缓存,确保在数据更新期间任何新生成的缓存也会被清除。

伪代码如下:

function delayDoubleDelete(key, updateStatement)// 第一步:删除Redis中的缓存Redis.delete(key)// 第二步:更新数据库Database.execute(updateStatement)// 第三步:执行短暂的延时,例如500毫秒sleep(500)// 第四步:再次删除Redis中的缓存Redis.delete(key)
end function

3.4. 使用WAL日志实现缓存同步(本文重点!!)

3.4.1. 使用WAL日志实现更新缓存的优势

在实际开发中,使用WAL(Write-Ahead Logging)日志同步Redis和PostgreSQL是一种非常高效的策略。相比于传统的缓存更新方法,WAL日志带来了许多优势:

1. 解耦操作,降低延迟

通过WAL日志,我们可以将数据库更新和缓存更新解耦。这意味着数据库操作可以独立完成,而不需要等待缓存更新,从而显著降低了响应时间,提升了系统的整体性能。

2. 确保数据一致性

WAL日志记录了数据库的所有变更操作,确保缓存更新时可以准确地反映数据库的最新状态。这种方式大大提高了数据的一致性,尤其是在高并发环境下,能够有效减少数据冲突和不一致的情况。

3. 增强容错和恢复能力

WAL日志具备出色的容错能力。即使系统发生故障,我们也可以通过重放WAL日志来恢复缓存中的数据,从而保证缓存与数据库的同步一致,快速恢复系统的正常运行。

4. 灵活性强

使用WAL日志还可以实现更加灵活的缓存更新策略。例如,你可以根据业务需求选择性地更新缓存,或者批量处理多条日志记录后再更新缓存,从而进一步优化系统性能。

总体来说,WAL日志为Redis和PostgreSQL的同步提供了一种高效、可靠的解决方案,特别适合需要高数据一致性和快速恢复能力的应用场景。

3.4.2. 基于Spring Boot实现缓存同步

在 Spring Boot 和 Java 环境中,实现 PostgreSQL 与Redis数据同步的技术流程如下:

在上述流程中:

  1. Java 生产者 监听 PostgreSQL 的 WAL 日志,捕获数据库变更。
  2. 如果存在数据更新,这些更新会被发送到 RabbitMQ。
  3. RabbitMQ 作为消息中间件,负责传递这些更新消息。
  4. Java 消费者 从 RabbitMQ 接收消息,解析这些消息,并据此在 Redis 中执行相应的 CRUD 操作。

这样的设计确保了 PostgreSQL 数据库的变更能够实时同步到 Redis 缓存中,从而维持两者之间的数据一致性。

本文仅展示WAL监听到同步Redis的核心步骤及代码(省去RabbitMQ),如对RabbitMQ构建感兴趣的jym可移步(【架构系列】RabbitMQ应用场景及在实际项目中如何搭建可靠的RabbitMQ架构体系-CSDN博客)。

步骤如下:

1. 修改 postgresql.conf 配置文件

wal_level = logical
max_replication_slots = 4
max_wal_senders = 4

2. 修改pg_hba.conf配置文件

# 允许复制连接
host    replication     all             192.168.100.0/24        md5

 这里,192.168.100.0/24应替换为实际的客户端IP地址或网段,md5表示使用密码认证。

3. 重启PostgreSQL服务

# 使用适当的命令重启PostgreSQL,这取决于你的操作系统和安装方式
sudo systemctl restart postgresql
# 或者
sudo service postgresql restart

 4. 编写监听WAL代码

package com.xfc.pg2redis.service;import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.stereotype.Service;import javax.annotation.PostConstruct;
import java.util.List;@Service
public class DatabaseChangeService {@Autowiredprivate JdbcTemplate jdbcTemplate;private static final String SLOT_NAME = "my_replication_slot";@Autowiredprivate ChangeLogProcessor changeLogProcessor;@PostConstructpublic void startListening() {new Thread(this::initializeReplicationSlot).start();}private void initializeReplicationSlot() {if (!isSlotExists(SLOT_NAME)) {createReplicationSlot(SLOT_NAME);}listenToReplicationSlot();}private boolean isSlotExists(String slotName) {String query = "SELECT COUNT(*) FROM pg_replication_slots WHERE slot_name = ?";Integer count = jdbcTemplate.queryForObject(query, new Object[]{slotName}, Integer.class);return count != null && count > 0;}private void createReplicationSlot(String slotName) {String query = "SELECT pg_create_logical_replication_slot(?, 'test_decoding')";jdbcTemplate.update(query, slotName);}public void listenToReplicationSlot() {String query = "SELECT data FROM pg_logical_slot_get_changes(?, NULL, NULL)";while (true) {List<String> changes = jdbcTemplate.queryForList(query, new Object[]{SLOT_NAME}, String.class);for (String change : changes) {
//                processChange(change);changeLogProcessor.processChangeLog(change);}try {Thread.sleep(10000);  // 暂停一段时间再次检查} catch (InterruptedException e) {Thread.currentThread().interrupt();}}}//    private void processChange(String change) {
//        // 处理数据库变更逻辑
//        System.out.println("Detected change: " + change);
//    }
}

 这段代码是一个 Spring Boot 服务类,它使用 JdbcTemplate 从 PostgreSQL 的逻辑复制槽中获取变更数据。服务启动时检查并创建逻辑复制槽,然后启动一个线程不断监听变更,处理变更数据并调用 ChangeLogProcessor 进行进一步处理。

5. 编写解析WAL日志代码

package com.xfc.pg2redis.service;import com.xfc.pg2redis.utils.RedisUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Service;@Service
public class ChangeLogProcessor {@Autowiredprivate RedisUtils redisUtils;public void processChangeLog(String changeLog) {// 解析日志内容String operation = extractOperation(changeLog);String id = extractId(changeLog);String data = extractData(changeLog);switch (operation) {case "INSERT":// 将数据插入到 RedisredisUtils.set(id, data);System.out.println("Inserted into Redis: " + id + " => " + data);break;case "UPDATE":// 更新 Redis 中的数据redisUtils.set(id, data);System.out.println("Updated in Redis: " + id + " => " + data);break;case "DELETE":// 从 Redis 中删除数据redisUtils.del(id);System.out.println("Deleted from Redis: " + id);break;default:System.out.println("Unknown operation: " + operation);}}private String extractOperation(String changeLog) {if (changeLog.startsWith("INSERT:")) {return "INSERT";} else if (changeLog.startsWith("UPDATE:")) {return "UPDATE";} else if (changeLog.startsWith("DELETE:")) {return "DELETE";}return "UNKNOWN";}private String extractId(String changeLog) {// 简单的正则表达式提取 idString idPattern = "id\\[character varying\\]:'(.*?)'";java.util.regex.Pattern pattern = java.util.regex.Pattern.compile(idPattern);java.util.regex.Matcher matcher = pattern.matcher(changeLog);return matcher.find() ? matcher.group(1) : "";}private String extractData(String changeLog) {// 提取数据内容,这里假设数据是以 `name`, `price`, `num` 字段组成// 这里只是一个简单的提取示例,根据实际情况调整String dataPattern = "name\\[character varying\\]:'(.*?)' price\\[bigint\\]:(\\d+) num\\[integer\\]:(\\d+)";java.util.regex.Pattern pattern = java.util.regex.Pattern.compile(dataPattern);java.util.regex.Matcher matcher = pattern.matcher(changeLog);if (matcher.find()) {return String.format("{\"name\":\"%s\", \"price\":%s, \"num\":%s}",matcher.group(1), matcher.group(2), matcher.group(3));}return "";}
}

上述代码定义了一个名为 ChangeLogProcessor 的服务类,用于处理 PostgreSQL 生成的变更日志,并根据这些日志更新 Redis 缓存。类中的 processChangeLog 方法接收一个变更日志字符串,通过 extractOperationextractIdextractData 方法提取操作类型、数据标识符(ID)和数据内容。根据操作类型(插入、更新、删除),该方法会调用 RedisUtils 类的方法来同步 Redis 缓存,确保 Redis 中的数据与 PostgreSQL 数据库保持一致。这样的处理流程有效支持了数据在不同存储系统间的一致性维护。 

3.4.3. 效果测试

1. 新增数据

调用新增数据接口:

 日志打印:


 redis效果:

2. 编辑数据

 调用编辑数据接口:


 日志打印:


 redis效果:

3.5. 源码地址:

xfc-fdw-cloud: 公共解决方案

4. 结语

通过本文的讨论,我们深入探讨了如何使用WAL日志实现PostgreSQL与Redis之间的高效数据同步。在现代应用程序中,确保数据库和缓存之间的数据一致性至关重要,而WAL日志提供了一种可靠的解决方案,既能保证高性能,又能确保数据的一致性。通过实际代码示例和流程解析,我们展示了如何在Spring Boot环境中实现这一机制,并验证了其在实际应用中的有效性。若本文对你有帮助。别忘记点点关注哦~

这篇关于数据同步的艺术:探索PostgreSQL和Redis的一致性策略的文章就介绍到这儿,希望我们推荐的文章对编程师们有所帮助!



http://www.chinasem.cn/article/1127516

相关文章

大模型研发全揭秘:客服工单数据标注的完整攻略

在人工智能(AI)领域,数据标注是模型训练过程中至关重要的一步。无论你是新手还是有经验的从业者,掌握数据标注的技术细节和常见问题的解决方案都能为你的AI项目增添不少价值。在电信运营商的客服系统中,工单数据是客户问题和解决方案的重要记录。通过对这些工单数据进行有效标注,不仅能够帮助提升客服自动化系统的智能化水平,还能优化客户服务流程,提高客户满意度。本文将详细介绍如何在电信运营商客服工单的背景下进行

基于MySQL Binlog的Elasticsearch数据同步实践

一、为什么要做 随着马蜂窝的逐渐发展,我们的业务数据越来越多,单纯使用 MySQL 已经不能满足我们的数据查询需求,例如对于商品、订单等数据的多维度检索。 使用 Elasticsearch 存储业务数据可以很好的解决我们业务中的搜索需求。而数据进行异构存储后,随之而来的就是数据同步的问题。 二、现有方法及问题 对于数据同步,我们目前的解决方案是建立数据中间表。把需要检索的业务数据,统一放到一张M

服务器集群同步时间手记

1.时间服务器配置(必须root用户) (1)检查ntp是否安装 [root@node1 桌面]# rpm -qa|grep ntpntp-4.2.6p5-10.el6.centos.x86_64fontpackages-filesystem-1.41-1.1.el6.noarchntpdate-4.2.6p5-10.el6.centos.x86_64 (2)修改ntp配置文件 [r

关于数据埋点,你需要了解这些基本知识

产品汪每天都在和数据打交道,你知道数据来自哪里吗? 移动app端内的用户行为数据大多来自埋点,了解一些埋点知识,能和数据分析师、技术侃大山,参与到前期的数据采集,更重要是让最终的埋点数据能为我所用,否则可怜巴巴等上几个月是常有的事。   埋点类型 根据埋点方式,可以区分为: 手动埋点半自动埋点全自动埋点 秉承“任何事物都有两面性”的道理:自动程度高的,能解决通用统计,便于统一化管理,但个性化定

使用SecondaryNameNode恢复NameNode的数据

1)需求: NameNode进程挂了并且存储的数据也丢失了,如何恢复NameNode 此种方式恢复的数据可能存在小部分数据的丢失。 2)故障模拟 (1)kill -9 NameNode进程 [lytfly@hadoop102 current]$ kill -9 19886 (2)删除NameNode存储的数据(/opt/module/hadoop-3.1.4/data/tmp/dfs/na

异构存储(冷热数据分离)

异构存储主要解决不同的数据,存储在不同类型的硬盘中,达到最佳性能的问题。 异构存储Shell操作 (1)查看当前有哪些存储策略可以用 [lytfly@hadoop102 hadoop-3.1.4]$ hdfs storagepolicies -listPolicies (2)为指定路径(数据存储目录)设置指定的存储策略 hdfs storagepolicies -setStoragePo

Hadoop集群数据均衡之磁盘间数据均衡

生产环境,由于硬盘空间不足,往往需要增加一块硬盘。刚加载的硬盘没有数据时,可以执行磁盘数据均衡命令。(Hadoop3.x新特性) plan后面带的节点的名字必须是已经存在的,并且是需要均衡的节点。 如果节点不存在,会报如下错误: 如果节点只有一个硬盘的话,不会创建均衡计划: (1)生成均衡计划 hdfs diskbalancer -plan hadoop102 (2)执行均衡计划 hd

深入探索协同过滤:从原理到推荐模块案例

文章目录 前言一、协同过滤1. 基于用户的协同过滤(UserCF)2. 基于物品的协同过滤(ItemCF)3. 相似度计算方法 二、相似度计算方法1. 欧氏距离2. 皮尔逊相关系数3. 杰卡德相似系数4. 余弦相似度 三、推荐模块案例1.基于文章的协同过滤推荐功能2.基于用户的协同过滤推荐功能 前言     在信息过载的时代,推荐系统成为连接用户与内容的桥梁。本文聚焦于

【Prometheus】PromQL向量匹配实现不同标签的向量数据进行运算

✨✨ 欢迎大家来到景天科技苑✨✨ 🎈🎈 养成好习惯,先赞后看哦~🎈🎈 🏆 作者简介:景天科技苑 🏆《头衔》:大厂架构师,华为云开发者社区专家博主,阿里云开发者社区专家博主,CSDN全栈领域优质创作者,掘金优秀博主,51CTO博客专家等。 🏆《博客》:Python全栈,前后端开发,小程序开发,人工智能,js逆向,App逆向,网络系统安全,数据分析,Django,fastapi

在JS中的设计模式的单例模式、策略模式、代理模式、原型模式浅讲

1. 单例模式(Singleton Pattern) 确保一个类只有一个实例,并提供一个全局访问点。 示例代码: class Singleton {constructor() {if (Singleton.instance) {return Singleton.instance;}Singleton.instance = this;this.data = [];}addData(value)