利用redissyncer实现数据双向同步

不知不觉【数据迁移专题】已经进行了两期，在先前《跨越异构鸿沟，Redis 迁移同步过程中的挑战与解决方案》和《在线数据迁移，数字化时代的必修课》中，我们为大家介绍了数据迁移挑战与技术选型，...

京东云开发者

3717人浏览 · 2021-06-15 17:13:25

京东云开发者 · 2021-06-15 17:13:25 发布

不知不觉【数据迁移专题】已经进行了两期，在先前《跨越异构鸿沟，Redis 迁移同步过程中的挑战与解决方案》和《在线数据迁移，数字化时代的必修课》中，我们为大家介绍了数据迁移挑战与技术选型，并详细分享京东云自研开源的RedisSyncer 项目。本篇是系列内容第三篇，我们来聊一聊如何用RedisSyncer实现数据双向同步。

redissyncer简介

RedisSyncer是京东云自研的redis多任务同步中间件工具集，应用于redis单实例及集群同步。该工具集包括：

redis 同步服务引擎 redissyncer-server
redissycner 客户端 redissyncer-cli
redis 数据校验工具 redissycner-compare
基于docker-compse的一体化部署方案 redissyncer

目前在github开源：

https://github.com/TraceNature/redissyncer-server

缓存同步的定义及必要性

双向同步是指在两个实例都有存量数据和写流量的情况下进行两实例同步，最终达到两实例数据动态一致的过程
缓存数据全局可读，防止缓存击穿
保证缓存命中率，为数据库减压
当单一数据中心发生故障时，保证数据在另一中心完全可见

双向同步的操作难度与冷启动问题

原生redis同步无法区分缓存数据来源
由于redis本身没有实例标识(类似mysql的GTID),在双向同步时形成数据回环
redis环状缓冲区覆盖后，数据混淆且难于清理

基于数据冲销的双向同步方案

利用数据冲销的方式破除数据写入环。该方案的必要条件是，同步的实例或集群写入的key无冲突，即在数据中心A写入的key，不会同一时间在B中心写入相异值。

假设我们有两个redis实例redis1和redis2，再分别定义两个冲销池set1和set2，记录key及同步次数。

启动redis1->redis2的全量同步任务
启动redis1->redis2增量任务 2-1。增量任务先在set1做冲销(set1中存在的数据删除并丢弃同步) 2-2. 未被冲销数据同步到redis2
启动redis2->redis1的全量任务，此全量同步数据一定会作为增量形成回环，所以要先写入set1再写入redis1，以便数据作为增量回环同步到redis2时利用set1冲销 3-1，写入set1 3-2，写入redis1
启动redis2->redis1增量任务，增量任务先在set2做冲销(set2中存在的数据删除并丢弃同步)，增量任务先写入set1在写入redis1避免循环复制 4-1。通过set2冲销数据 4-2，数据写入set1 4-3，数据写入redis1
改变redis1->redis2增量任务行为，增量任务先在set1做冲销(set1中存在的数据删除并丢弃同步)，未冲销数据先写入set2再同步到redis2 5-1 写入set2 5-2 写入redis2

为什么增加第五步改变redis1->redis2增量任务行为呢？因为在第四步完成时set2中并没有从redis1->redis2的增量数据，这会造成从redis1->redis2的增量数据会转换成redis2->redis1增量数据且在本地无法被冲销，只有数据进入set1且被写入redis1后再次作为增量数据向redis2同步时才会被冲销，增加了网络开销同时redis1也增加了一次写入负载。

数据冲销方式及其缺陷

数据冲销方式需要在每次发送数据前对数据进行缓冲，正常情况下缓冲内存占用不大。但当网络阻塞或由于网络不畅无法冲销数据时，会造成缓冲区暴增导致OOM
数据冲销方式带来的冷启动问题
- 当任务异常中断且redis offset被覆盖的情况下，因为数据矫正依据缺失，需要重建缓存
- 若采用数据持久化的方式先持久化后发送的方式，那么在冲销过程中会大大降低同步效率

数据双写，看似美好其实坑多多

业务双写是最符合人类直觉的双向方案，同一份数据写入两个数据中心以保障数据冗余。但是在实际操作中会遇到数据写入顺序问题。

双写方案中的数据顺序问题

并发环境中同时写入同一个key的情况下，并不能保障key写入redis的顺序。造成key的结果不一致。

通过统一队列解决顺序问题
网络中断导致数据缺失
强一致性会导致单机房不可用的情况下写操作全局不可用，并需要在数据在某一次提交不完全成功的情况下提供回滚机制、及数据补偿机制
队列带来写效率损失，redis失去作为缓存层的意义

双读方案

每个数据中心建立redis 读写实例与只读实例
读写数据中的落地数据通过redissyncer同步到对端只读实例
应用读数据时先读取只读实例若有数据返回则返回；若无数据返回则读取读写实例
双读方案的限制条件
- key的生成在全局具有唯一性既两个中心不出现重复的key
- 避免incr 、 lpush 等非幂等操作
- 由于网络抖动可能造成数据流中断，尽管redissyncer以及对非幂等命令做了处理，但是极端情况仍然可能造成数据不准确影响业务

双读方案实践模拟

环境列表

主机名	IP地址	部署软件或工具
az_a1	10.0.0.110	redis5.0
az_a1	10.0.0.110	redissyncer
az_a2	10.0.0.111	redis5.0
az_b1	10.0.0.112	redis5.0
az_b1	10.0.0.112	redissyncer
az_b2	10.0.0.113	redis5.0

az_a1 代表 a 中心的 redis RW 实例；az_a2 代表 a 中心 redis RO 实例；az_b1 代表 b 中心 redis RW 实例；az_b2 代表 b 中心 redis RO 实例
分别在 az_a1、az_b1上部署 redissyncer 用于同步到对端数据中心
通过 redisdual 模拟双读客户端

实施细则

部署 redis 详见 redis 部署文档，这里不累述
部署redissyncer（docker方式）; az_a1、az_b1 上执行

clone redissyncer 项目

1git clone https://github.com/TraceNature/redissyncer.git
2cd redissyncer
3docker-compose up -d

部署 redissyncer-cli 用于与服务器通讯

下载客户端程序

1wget https://github.com/TraceNature/redissyncer-cli/releases/download/v0.   1.0/redissyncer-cli-0.1.0-linux-amd64.tar.gz
2
3tar zxvf redissyncer-cli-0.1.0-linux-amd64.tar.gz

配置.config.yaml 文件

1# redissyncer-server 访问地址及端口
2syncserver: http://127.0.0.1:8080
3# 访问 redissyncer-server 的 token。可以通过 redissyncer-cli login 命令获得。默认用户名 admin 默认密码 123456.完整命令 redissyncer-cli login admin 123456
4token: 379F5E2BD55A4608B6A7557F0583CFC5

az_a1 配置同步任务同步到 az_b2

编辑任务文件 synctask/a1_to_b2.json

 1{
 2"sourcePassword": "redistest0102",
 3"sourceRedisAddress": "10.0.0.110:16375",
 4"targetRedisAddress": "10.0.0.113:16375",
 5"targetPassword": "redistest0102",
 6"taskName": "a1_to_b2",
 7"targetRedisVersion": 5.0,
 8"autostart": true,
 9"afresh": true,
10"batchSize": 100
11}

启动任务

1redissyncer-cli-0.1.0-linux-amd64 -i
2redissyncer-cli> task create source synctask/a1_to_b2.json;

az_b1 配置同步任务同步到 az_a2

编辑任务文件 synctask/b1_to_a2.json

 1{
 2  "sourcePassword": "redistest0102",
 3  "sourceRedisAddress": "10.0.0.112:16375",
 4  "targetRedisAddress": "10.0.0.111:16375",
 5  "targetPassword": "redistest0102",
 6  "taskName": "b1_to_a2",
 7  "targetRedisVersion": 5.0,
 8  "autostart": true,
 9  "afresh": true,
10  "batchSize": 100
11}

启动任务

1redissyncer-cli-0.1.0-linux-amd64 -i
2redissyncer-cli> task create source synctask/b1_to_a2.json;

通过redisdual 模拟redis 双读
克隆 https://github.com/TraceNature/redisdual.git 项目自行编译

config.yaml 文件参数详解

 1# 日志配置不需改动
 2zap:
 3level: 'debug'
 4format: 'console'
 5prefix: '[redisdual]'
 6director: 'log'
 7link-name: 'latest_log'
 8show-line: true
 9encode-level: 'LowercaseColorLevelEncoder'
10#  stacktrace-key: 'stacktrace'
11log-in-console: true
12
13# 执行时间间隔，单位毫秒，可以控制每次执行的间隔时长，便于观察日志
14execinterval: 1
15
16# 循环执行最大步长，当大于步长时归零；当达到步长时，重新执行写入和双读操作
17loopstep: 30
18
19# 本地key前缀，区分本地写入key和remote端写入的key
20localkeyprefix: a
21remotekeyprefix: b
22
23# redis 读写实例
24redisrw:
25db: 0
26addr: '114.67.76.82:16375'
27password: 'redistest0102'
28# redis 只读实例
29redisro:
30db: 0
31addr: '114.67.120.120:16375'
32password: 'redistest0102'

主要代码分析

redisdual/cmd/start.go；func dual(rw *redis.Client, ro *redis.Client, key string) 函数.双读的主要逻辑，先读RO实例，有返回值输出，无返回值读RW库，有返回值输出，无返回值结束查询

1func dual(rw *redis.Client, ro *redis.Client, key string) {
 2  roResult, err := ro.Get(key).Result()
 3
 4  if err == nil && roResult != "" {
 5      global.RSPLog.Sugar().Infof("Get key %s from redisro result is:%s ",     key, roResult)
 6      return
 7  }
 8
 9  rwResult, err := rw.Get(key).Result()
10  if err != nil || rwResult == "" {
11      global.RSPLog.Sugar().Infof("key %s no result return!", key)
12      return
13  }
14
15  global.RSPLog.Sugar().Infof("Get key %s from redisrw result is: %s ",    key, rwResult)
16
17}

redisdual/cmd/start.go；func startServer() 函数。启动服务，定时执行RW实例写入。并执行双读操作

 1// -d 后台启动
 2if global.RSPViper.GetBool("daemon") {
 3cmd, err := background()
 4if err != nil {
 5panic(err)
 6}
 7
 8//根据返回值区分父进程子进程
 9if cmd != nil { //父进程
10fmt.Println("PPID: ", os.Getpid(), "; PID:", cmd.Process.Pid, "; Operating parameters: ", os.Args)
11return //父进程退出
12} else { //子进程
13fmt.Println("PID: ", os.Getpid(), "; Operating parameters: ", os.Args)
14}
15}
16
17global.RSPLog = core.Zap()
18global.RSPLog.Info("server start ... ")
19
20pidMap := make(map[string]int)
21// 记录pid
22pid := syscall.Getpid()
23pidMap["pid"] = pid
24
25pidYaml, _ := yaml.Marshal(pidMap)
26dir, err := filepath.Abs(filepath.Dir(os.Args[0]))
27if err != nil {
28panic(err)
29}
30
31if err := ioutil.WriteFile(dir+"/pid", pidYaml, 0664); err != nil {
32global.RSPLog.Sugar().Error(err)
33panic(err)
34}
35global.RSPLog.Sugar().Infof("Actual pid is %d", pid)
36
37//redis 读写实例
38redisRW := GetRedisRW()
39
40//redis 只读实例
41redisRO := GetRedisRO()
42
43//清理RW
44redisRW.FlushAll()
45
46global.RSPLog.Sugar().Info("execinterval:", global.RSPViper.GetInt("execinterval"))
47loopstep := global.RSPViper.GetInt("loopstep")
48i := 0
49for {
50if i > loopstep {
51i = 0
52}
53key := global.RSPViper.GetString("localkeyprefix") + "_key" + strconv.Itoa(i)
54redisRW.Set(key, key+"_"+strconv.FormatInt(time.Now().UnixNano(), 10), 3600*time.Second)
55dual(redisRW, redisRO, global.RSPViper.GetString("localkeyprefix")+"_key"+strconv.Itoa(i))
56dual(redisRW, redisRO, global.RSPViper.GetString("remotekeyprefix")+"_key"+strconv.Itoa(i))
57i++
58time.Sleep(time.Duration(global.RSPViper.GetInt("execinterval")) * time.Millisecond)
59}