一.canal服务端

canal 是阿里巴巴 MySQL 数据库 Binlog 的增量订阅 & 消费组件。基于数据库增量日志解析，提供增量数据订阅 & 消费。

二.基于docker搭建

在docker环境中

1.执行 docker search canal 获取到需要下载的canal镜像

2.执行docker pull 选择下载的canal镜像

3.执行run

docker run \
--name canal \
-p 11111:11111 \
-d canal/canal-server:latest

4.获取配置文件到宿主机

进入canal
docker exec -it canal /bin/bash
//查看对应的conf文件在那个目录
复制配置文件到宿主机指定目录
docker cp /home/admin/canal-server/conf /docker/canal/config

3.在自己的liunx系统创建需要挂载的文件夹 我这边创建了config和logs文件

docker run \
--name canal \
-p 11111:11111 \
-v /docker/canal/config:/home/admin/canal-server/conf \
-v /docker/canal/logs:/home/admin/canal-server/logs \
-d canal/canal-server:latest

4.修改canal.properties配置

重要配置：

# canal端口
canal.port = 11111

# 可以支持tcp, kafka, rocketMQ, rabbitMQ, pulsarMQ
canal.serverMode = tcp

#实例名称，可以配置多个，配置多个会生产多个文件夹
canal.destinations = example

5.修改对应实力instance.properties配置

重要配置：

# 配置你连接的数据库地址
canal.instance.master.address=127.0.0.1:3306

# 配置数据库的username/password
canal.instance.dbUsername=root
canal.instance.dbPassword=123456

# table regex 监听的数据库规则
#.your_database 表示你要同步的数据库名，\\.表示匹配一个点号（.）的转义字符，.在正则表达式中表示匹配任意字符，因此需要转义表示匹配点号。.*表示匹配所有以 your_database.开头的字符串。
canal.instance.filter.regex=your_database\\..*

三.开启binglog日志

 1.使用canal需要mysql开启binlog日志

[mysqld]
#设置服务器id(确保唯一)
server-id=666
#设置日志个数默认ROW 根据需求选择
binlog_format=ROW
#开启binlog日志（mysql-bin 日志名称）
log-bin=mysql-bin
#设置需要复制的数据库，默认复制全部
#binlog-do-db=mydb

2.重启mysql服务

systemctl restart mysqld

3.查看状态（如果不为空则配置成功）

SHOW MASTER STATUS;

四.集成canal

1.引入jar

 <!-- canal 同步双写-->
        <dependency>
            <groupId>top.javatool</groupId>
            <artifactId>canal-spring-boot-starter</artifactId>
            <version>1.2.1-RELEASE</version>
        </dependency>

2.yml配置

canal:
  server: 192.168.1.1:11111
  destination: example

3.代码

import lombok.Data;

import javax.persistence.Column;
import javax.persistence.Id;


@Data
public class UserBean {

    @Id
    public Integer id;
    @Column(name = "name")
    public String name;
    @Column(name = "age")
    public Integer age;

    @Override
    public String toString() {
        return "UserBean{" + "id=" + id + ", name='" + name + '\'' + ", age=" + age + '}';
    }
}

import com.pw.canal.bean.UserBean;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;
import top.javatool.canal.client.annotation.CanalTable;
import top.javatool.canal.client.handler.EntryHandler;
@Slf4j
@CanalTable("sc_user")
@Component
public class UserBeanHandler implements EntryHandler<UserBean> {

    @Override
    public void insert(UserBean userBean ) {
      log.info("对象{}",userBean );
       //添加本地缓存
        //添加redis缓存
    }

    @Override
    public void update(UserBean before, UserBean after) {
        log.info("对象{}",before);
        log.info("对象{}",after);
       //添加本地缓存
        //添加redis缓存
    }

    @Override
    public void delete(UserBean userBean ) {
        log.info("对象{}",userBean );
       //添加本地缓存
        //添加redis缓存
    }
}

五.注意

1.如果控制台又打印日志,但是没有进入hardler,注意对象的转换字段类型是否匹配.

2.mysql配置文件中,是否设置了 binlog_format=ROW