第一步开启 Binlog 写入功能 修改配置文件my.cnf

log-bin=mysql-bin # 开启 binlog
binlog-format=ROW # 选择 ROW 模式
server_id=1 # 配置 MySQL replaction 需要定义，不要和 canal 的 slaveId 重复

show variables like '%log_bin%' #查看是否开启 binlog

第二步 创建mysql canal用户

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

第三步 docker拉取 canal镜像

docker pull canal/canal-server:v1.1.4

第四步 docker启动canal

docker run -p 11111:11111 --name canal -d canal/canal-server:v1.1.4

其它方式 如docker-compose 安装

version: '3'
services:
  canal:
    image: canal/canal-server:v1.1.4
    container_name: canalserver01
    restart: always
    ports:
   - 11111:11111
     lumes: 
        - /root/canalserver/canal.properties:/home/admin/canal-server/conf/canal.properties
          /root/canalserver/logs:/home/admin/canal-server/logs
             - /root/canalserver/instance.properties:/home/admin/canal-server/conf/example/instance.properties

第五步进入容器

docker exec -it canal bash

第6步home/admin/canal-server/conf/example目录下编辑instance.properties文件 如下

canal.instance.mysql.slaveId=23
#配置mysql的地址和端口
canal.instance.master.address=服务器ip:3306
#配置前面创建的专用于canal的mysql账号和密码
canal.instance.dbUsername=canal
canal.instance.dbPassword=canal

整合springboot

导入maven依赖

	  
		  top.javatool
		  canal-spring-boot-starter
		  1.2.1-RELEASE
	  
	  
		  org.springframework.boot
		  spring-boot-autoconfigure
	  

配置yml

#canal 配置
canal:
  server: 服务地址:11111
  destination: example

案例

接收 增删改对象

import com.fasterxml.jackson.annotation.JsonFormat;
import io.swagger.annotations.ApiModelProperty;
import lombok.Data;

import javax.persistence.*;
import java.io.Serializable;



@Data
@Table(name = "hospital_bed")
public class HospitalBed implements Serializable {
    
    @Id
    @GeneratedValue(strategy = GenerationType.IDENTITY)
    @Column(name = "id")
    private Long id;

    

    @Column(name = "department_id")
    private String departmentId;

    
    @JsonFormat(shape = JsonFormat.Shape.STRING)
    @Column(name = "patient_basic_id")
    private String patientBasicId;

    

    @Column(name = "bed_type")
    private String bedType;

    
    @Column(name = "bed_number")
    private String bedNumber;

    

    @Column(name = "bed_status")
    private String bedStatus;


    @Column(name = "is_del")
    private String isDel;

    

    @Column(name = "created_by")
    private String createdBy;

    
    @Column(name = "created_time")
    private String createdTime;

    
    @Column(name = "updated_by")
    private String updatedBy;

    
    @Column(name = "updated_time")
    private String updatedTime;

}

import com.hyb.yfk.canal.dto.HospitalBed;
import com.hyb.yfk.config.BulletinBoardWebSocketServer;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import top.javatool.canal.client.annotation.CanalTable;
import top.javatool.canal.client.handler.EntryHandler;



@Slf4j
@Component
@CanalTable("hospital_bed")
public class HospitalBedHandler implements EntryHandler {
    @Override
    public void insert(HospitalBed bed) {
        log.info("bed--添加: message  {}", bed);
        
    }

    @Override
    public void update(HospitalBed beforeBed, HospitalBed afterBed) {
        log.info("bed--修改前: {} ", beforeBed);
        log.info("bed--修改后: {}", afterBed);
       
    }

    @Override
    public void delete(HospitalBed bed) {
        log.info("bed--删除:  {}", bed);
       

    }
}

解读注解

@Table	对应数据库表名
@Column	对应数据库字段名称
@CanalTable	需要监听的表名

踩坑

启动失败查看是否有该注解   

@AllArgsConstructor

关闭canal日志

在logback-spring.xml 添加以下 即可

第一次写作 如有错误 请指正 谢谢大家!

你们的点赞就是我更新的动力

禁止转载! 如需转载请和我联系