canal使用说明：MySQL、Redis实时数据同步_canal基于mysql binlog日志同步的。但是有服务器系统有要求

1. canal简介

canal是阿里开源的数据同步工具，基于bin log可以将数据库同步到其他各类数据库中，目标数据库支持mysql,postgresql,oracle,redis,MQ,ES等

canal分成服务端deployer和客户端adapter，我们可以部署多个，同时为了方便管理还提供了一个管理端admin

canal的数据同步流程如下图所示

因为目前canal还不能直接通过配置就实现对redis的数据同步，因此我们需要自定义一下canal客户端，通过服务端将数据同步到客户端后，由客户端自定义操作同步到redis

2、canal配置（MySQL数据库用的是204服务器上的，canal安装在208服务器上）

第一步：因为canal是监控MySQL中的binlog日志来完成同步工作，所以我们需要开启binlog日志功能，由于咱们项目用的MySQL8以上的版本，binlog默认开启，所以可以不管。

第二步：通过show variables like ‘%binlog_format%’; 查看当前binlog，值为ROW就可以了

第三步：源数据库创建一个canal账号，并且设置slave,dump权限

CREATE USER canal IDENTIFIED BY 'canal';  
GRANT SELECT, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO 'canal'@'%';
-- GRANT ALL PRIVILEGES ON *.* TO 'canal'@'%' ;
FLUSH PRIVILEGES;

1
2
3
4
5

第四步：因为mysql8.0.3后身份检验方式为caching_sha2_password，但canal使用的是mysql_native_password，因此需要设置检验方式（如果该版本之前的可跳过），否则会报错IOException: caching_sha2_password Auth failed

ALTER USER 'canal'@'%' IDENTIFIED WITH mysql_native_password BY 'canal';
select host,user,plugin from mysql.user ;
1
2

第五步：在208服务器上拉取canal，输入以下命令并解压到/tmp/canal文件夹下

wget https://github.com/alibaba/canal/releases/download/canal-1.1.6/canal.deployer-1.1.6.tar.gz
1

进入到canal文件夹，然后进入conf文件夹找到example文件夹，复制example文件夹并重命名为redis

cd conf
# 复制example实例配置
cp -R example redis
1
2
3

然后修改redis文件夹中的instance.properties配置文件

在此之前需要查看一下数据库binlog日志状态信息：

show master status;
1

然后进行instance.properties配置文件开始配置（说明和需要改动的地方都在图里）

想了解配置中其他配置信息的，可以参考下面的

配置完成后，启动canal客户端并查看日志，没报错就说明成功了

./bin/start.sh
cat logs/redis/redis.log
1
2

我在过程中也遇到了错误，如遇到，可以看看下面的解决办法：

Received error packet: errno = 1236, sqlstate = HY000 errmsg = Could
not find first log file name in binary log index file

第六步：创建canal应用端，配置需要同步到redis的地址（展示本地应用，具体的应用这几天会上传到云效上）

依赖引用：

<dependency>
  <groupId>top.javatool</groupId>
  <artifactId>canal-spring-boot-starter</artifactId>
  <version>1.2.1-RELEASE</version>
</dependency>
1
2
3
4
5

配置文件（application.yml）：

# 应用名称
spring:
  application:
    name: ns-service-support-canal
  # 数据同步到当前redis中
  redis:
    host: 127.0.0.1
    password: 123456
    database: 6

# 应用服务 WEB 访问端口
server:
  port: 8080

# canal服务端地址
canal:
  # 部署到208上，port默认为11111
  server: 192.168.xxx.xxx:11111
  # 实例名，与之前复制example之后那个文件夹名称一致
  destination: redis

# 设置canal消息日志打印级别
logging:
  level:
    top.javatool.canal.client: warn
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25

RedisConfig.java(json转换与其他微服务同步)

@Configuration
@AllArgsConstructor
public class RedisConfig {

    private RedisConnectionFactory factory;

    /**
     * jsonObject转换
     * @param connectionFactory
     * @return
     */
    @Bean
    @SuppressWarnings(value = { "unchecked", "rawtypes" })
    public RedisTemplate<Object, Object> redisTemplate(RedisConnectionFactory connectionFactory)
    {
        RedisTemplate<Object, Object> template = new RedisTemplate<>();
        template.setConnectionFactory(connectionFactory);

        FastJsonRedisSerializer serializer = new FastJsonRedisSerializer(Object.class);

        // 使用StringRedisSerializer来序列化和反序列化redis的key值
        template.setKeySerializer(new StringRedisSerializer());
        template.setValueSerializer(serializer);

        // Hash的key也采用StringRedisSerializer的序列化方式
        template.setHashKeySerializer(new StringRedisSerializer());
        template.setHashValueSerializer(serializer);

        template.afterPropertiesSet();
        return template;
    }

}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33

3、关键（如果选择直接使用当前项目，你需要在项目中增加实体类和handler两个）

关键的来了，需要同步哪张表，就需要创建对应的实体类（与其他微服务基本一致，需要添加的东西是，实体类头部增加@Table注解，标明对应的数据库表名，每个实体类参数添加@Column注解，名称对应数据库表参数），以user表为例

@Data
@Table(name = "user")
public class User implements Serializable {

    private static final long serialVersionUID = 1L;

    @ApiModelProperty("id")
    @NsGridField(title = "id")
    @Column(name="id")
    private Long userId;

    @ApiModelProperty("姓名")
    @NsGridField(title = "姓名")
    @Column(name="name")
    private String name;

    @ApiModelProperty("账号")
    @NsGridField(title = "账号")
    @Column(name="username")
    private String username;

    @ApiModelProperty("密码")
    @NsGridField(title = "密码")
    @Column(name="password")
    private String password;

    @ApiModelProperty("昵称")
    @NsGridField(title = "昵称")
    @Column(name="nickname")
    private String nickname;

    @ApiModelProperty("授权顾问昵称")
    @NsGridField(title = "授权顾问昵称")
    @Column(name="g_a_a_c_nickname")
    private String gAACNickname;

    @ApiModelProperty("头像地址")
    @NsGridField(title = "头像地址")
    @Column(name="avatar_url")
    private String avatarUrl;

    @ApiModelProperty("手机")
    @NsGridField(title = "手机")
    @Column(name="phone")
    private String phone;

    @ApiModelProperty("身份证号")
    @NsGridField(title = "身份证号")
    @Column(name="id_card")
    private String idCard;

    @ApiModelProperty("邮箱")
    @NsGridField(title = "邮箱")
    @Column(name="email")
    private String email;

    @ApiModelProperty("性别: 0男;1女;2未知")
    @NsGridField(title = "性别: 0男;1女;2未知")
    @Column(name="gender")
    private Integer gender;

    @ApiModelProperty("账号状态：0正常;1禁用")
    @NsGridField(title = "账号状态：0正常;1禁用")
    @Column(name="account_status")
    private Integer accountStatus;

    @ApiModelProperty("账号类型：0普通用户 1平台用户 ")
    @NsGridField(title = "账号类型：0普通用户 1平台用户 ")
    @Column(name="account_type")
    private Integer accountType;

    @ApiModelProperty("是否实名认证：0否 1是")
    @NsGridField(title = "是否实名认证：0否 1是")
    @Column(name="is_approve")
    private Integer isApprove;

    @ApiModelProperty("平台组织id")
    @NsGridField(title = "平台组织id")
    @Column(name="org_id")
    private Long orgId;

    @ApiModelProperty("备注")
    @NsGridField(title = "备注")
    @Column(name="remark")
    private String remark;

    @ApiModelProperty("乐观锁")
    @NsGridField(title = "乐观锁")
    @Column(name="revision")
    private Integer revision;

    @ApiModelProperty("排序")
    @NsGridField(title = "排序")
    @Column(name="sort_value")
    private Integer sortValue;

    @ApiModelProperty("状态")
    @NsGridField(title = "状态")
    @Column(name="status")
    private Boolean status;

    @ApiModelProperty("组织描述")
    @NsGridField(title = "组织描述")
    @Column(name="org_describe")
    private String orgDescribe;

    @Override
    public String toString() {
        return "User{" +
                "name='" + name + '\'' +
                ", username='" + username + '\'' +
                ", password='" + password + '\'' +
                ", nickname='" + nickname + '\'' +
                ", gAACNickname='" + gAACNickname + '\'' +
                ", avatarUrl='" + avatarUrl + '\'' +
                ", phone='" + phone + '\'' +
                ", idCard='" + idCard + '\'' +
                ", email='" + email + '\'' +
                ", gender=" + gender +
                ", accountStatus=" + accountStatus +
                ", accountType=" + accountType +
                ", isApprove=" + isApprove +
                ", orgId=" + orgId +
                ", remark='" + remark + '\'' +
                ", revision=" + revision +
                ", sortValue=" + sortValue +
                ", status=" + status +
                ", orgDescribe='" + orgDescribe + '\'' +
                '}';
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131

canal-spring-boot-starter包提供了EntryHandler类用于监控表数据更新，于是我们创建一个EntryHandler实现类，用于实现redis的增删改

/**
 * @author csf
 * @Description
 * @date 2023/1/30
 */
@CanalTable("user")
@Component
@AllArgsConstructor
@Slf4j
public class NmsUserHandler  implements EntryHandler<User> {

    private final RedisTemplate<Object,Object> redisTemplate;

    @Override
    public void insert(User nmsUser) {
        log.info("[新增]"+nmsUser.toString());
        redisTemplate.opsForValue().set("user:"+nmsUser.getUsername(),nmsUser);
        Object o = redisTemplate.opsForValue().get("user:ns0");
        System.out.println(o);
    }

    @Override
    public void update(User before, User after) {
        log.info("[更新]"+after.toString());
        redisTemplate.opsForValue().set("user:"+after.getUsername(),after);
        Object o = redisTemplate.opsForValue().get("user:snn7");
        System.out.println(o);
    }

    @Override
    public void delete(User nmsUser) {
        log.info("[删除]"+nmsUser.getUsername());
        redisTemplate.delete("user:"+nmsUser.getUsername());
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36

至此，配置完成，启动项目

测试：不管对数据进行增或删改，日志都能实时反馈，并且数据都能实时同步到redis中

日志

redis