# 準(zhǔn)備
git clone https://gitee.com/zhengqingya/docker-compose.git
cd docker-compose/Liunx

# 導(dǎo)入初始化SQL
Liunx/canal/canal_admin/canal_manager.sql

# 運(yùn)行  (tips:先修改配置文件信息)
docker-compose -f docker-compose-canal.yml -p canal up -d

# 可參考 https://github.com/alibaba/canal/wiki/QuickStart

version: '3'

# 網(wǎng)橋canal -> 方便相互通訊
networks:
  canal:

services:
  canal_admin:
    image: registry.cn-hangzhou.aliyuncs.com/zhengqing/canal-admin:v1.1.5      # 原鏡像`canal/canal-admin:v1.1.5`
    container_name: canal_admin               # 容器名為'canal_admin'
    restart: unless-stopped                   # 指定容器退出后的重啟策略為始終重啟，但是不考慮在Docker守護(hù)進(jìn)程啟動(dòng)時(shí)就已經(jīng)停止了的容器
    volumes: # 數(shù)據(jù)卷掛載路徑設(shè)置,將本機(jī)目錄映射到容器目錄
      - "./canal/canal-admin/bin/startup.sh:/home/admin/canal-admin/bin/startup.sh"
      - "./canal/canal-admin/logs:/home/admin/canal-admin/logs"
    environment: # 設(shè)置環(huán)境變量,相當(dāng)于docker run命令中的-e
      TZ: Asia/Shanghai
      LANG: en_US.UTF-8
      canal.adminUser: admin
      canal.adminPasswd: 123456
      spring.datasource.address: www.zhengqingya.com:3306
      spring.datasource.database: canal_manager
      spring.datasource.username: root
      spring.datasource.password: root
    ports:
      - "8089:8089"
    networks:
      - canal

  canal_server:
    image: registry.cn-hangzhou.aliyuncs.com/zhengqing/canal-server:v1.1.5      # 原鏡像`canal/canal-server:v1.1.5`
    container_name: canal_server              # 容器名為'canal_server'
    restart: unless-stopped                   # 指定容器退出后的重啟策略為始終重啟，但是不考慮在Docker守護(hù)進(jìn)程啟動(dòng)時(shí)就已經(jīng)停止了的容器
    volumes:                                  # 數(shù)據(jù)卷掛載路徑設(shè)置,將本機(jī)目錄映射到容器目錄
      - "./canal/canal-server/logs:/home/admin/canal-server/logs"
    environment:                              # 設(shè)置環(huán)境變量,相當(dāng)于docker run命令中的-e
      TZ: Asia/Shanghai
      LANG: en_US.UTF-8
      canal.register.ip: www.zhengqingya.com
      canal.admin.manager: canal_admin:8089
      canal.admin.port: 11110
      canal.admin.user: admin
      canal.admin.passwd: 6BB4837EB74329105EE4568DDA7DC67ED2CA2AD9
    ports:
      - "11110:11110"
      - "11111:11111"
      - "11112:11112"
    depends_on:
      - canal_admin
    links:
      - canal_admin
    networks:
      - canal

# canal admin config
#canal.admin.manager = 127.0.0.1:8089
canal.admin.port = 11110
canal.admin.user = admin
canal.admin.passwd = 6BB4837EB74329105EE4568DDA7DC67ED2CA2AD9

# tcp, kafka, rocketMQ, rabbitMQ
canal.serverMode = rabbitMQ

rabbitmq.host = www.zhengqingya.com:5672
rabbitmq.virtual.host = my_vhost
rabbitmq.exchange = canal.exchange
rabbitmq.username = admin
rabbitmq.password = admin

canal.instance.master.address=www.zhengqingya.com:3306
# username/password
canal.instance.dbUsername=canal
canal.instance.dbPassword=canal
canal.instance.connectionCharset = UTF-8

# table regex 只同步test數(shù)據(jù)庫(kù)下的t_user表
canal.instance.filter.regex=test\\.t_user
# table black regex
canal.instance.filter.black.regex=mysql\\.slave_.*

# mq config
canal.mq.topic=canal_routing_key

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-amqp</artifactId>
</dependency>

# RabbitMQ配置
spring:
  rabbitmq:
    addresses: www.zhengqingya.com:5672,www.zhengqingya.com:5673 # 指定client連接到的server的地址，多個(gè)以逗號(hào)分隔
    # 填寫自己安裝rabbitmq時(shí)設(shè)置的賬號(hào)密碼，默認(rèn)賬號(hào)密碼為`guest`
    username: admin
    password: admin
    virtual-host: my_vhost # 填寫自己的虛擬機(jī)名，對(duì)應(yīng)可查看 `127.0.0.1:15672/#/users` 下Admin中的`Can access virtual hosts`信息

@Slf4j
@Component
public class CanalRabbitMqListener {

    @RabbitListener(bindings = {
            @QueueBinding(
                    value = @Queue(value = MqConstant.CANAL_QUEUE, durable = "true"),
                    exchange = @Exchange(value = MqConstant.CANAL_EXCHANGE),
                    key = MqConstant.CANAL_ROUTING_KEY
            )
    })
    public void handleCanalDataChange(String message) {
        log.info("[canal] 接收消息: {}", JSON.toJSONString(message));
    }

}

public interface MqConstant {

    String CANAL_EXCHANGE = "canal.exchange";
    String CANAL_QUEUE = "canal_queue";
    String CANAL_ROUTING_KEY = "canal_routing_key";

}

<!-- https://mvnrepository.com/artifact/top.javatool/canal-spring-boot-starter -->
<dependency>
    <groupId>top.javatool</groupId>
    <artifactId>canal-spring-boot-starter</artifactId>
    <version>1.2.1-RELEASE</version>
</dependency>

canal:
  server: www.zhengqingya.com:11111
  destination: example # tips：canal-admin中Instance管理下需存在example實(shí)例配置

@Slf4j
@Component
@CanalTable(value = "t_user")
public class UserHandler implements EntryHandler<User> {

    @Override
    public void insert(User user) {
        log.info("insert message  {}", user);
    }

    @Override
    public void update(User before, User after) {
        log.info("update before {} ", before);
        log.info("update after {}", after);
    }

    @Override
    public void delete(User user) {
        log.info("delete  {}", user);
    }

}

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

    /**
     * 主鍵
     */
    @Id
    @GeneratedValue(strategy = GenerationType.IDENTITY)
    @Column(name = "user_id")
    private Integer userId;

    /**
     * 用戶名
     */
    @Column(name = "username")
    private String username;

    /**
     * 密碼
     */
    @Column(name = "password")
    private String password;

    /**
     * 性別
     */
    @Column(name = "sex")
    private Integer sex;

    /**
     * 備注
     */
    private String remark;

    /**
     * 時(shí)間
     */
    private Date date;

}