canal使用说明：MySQL、Redis实时数据同步

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;

第四步：因为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 ;

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

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

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

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

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

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

show master status;

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

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

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

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

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

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>

配置文件（application.yml）：

# 应用名称
spring:application:name: ns-service-support-canal# 数据同步到当前redis中redis:host: 127.0.0.1password: 123456database: 6# 应用服务 WEB 访问端口
server:port: 8080# canal服务端地址
canal:# 部署到208上，port默认为11111server: 192.168.xxx.xxx:11111# 实例名，与之前复制example之后那个文件夹名称一致destination: redis# 设置canal消息日志打印级别
logging:level:top.javatool.canal.client: warn

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;}}

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;@Overridepublic 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 + '\'' +'}';}
}

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;@Overridepublic 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);}@Overridepublic 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);}@Overridepublic void delete(User nmsUser) {log.info("[删除]"+nmsUser.getUsername());redisTemplate.delete("user:"+nmsUser.getUsername());}
}

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

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

日志

redis