基于docker-compose部署canal

canal由来及相关介绍

canal [kə'næl]，译意为水道/管道/沟渠，主要用途是基于 MySQL 数据库增量日志解析，提供增量数据订阅和消费

早期阿里巴巴因为杭州和美国双机房部署，存在跨机房同步的业务需求，实现方式主要是基于业务 trigger 获取增量变更。从 2010 年开始，业务逐步尝试数据库日志解析获取增量变更进行同步，由此衍生出了大量的数据库增量订阅和消费业务。

详见>> Github：https://github.com/alibaba/canal

canal部署前Mysql相关设定

对于自建 MySQL , 需要先开启 Binlog 写入功能，配置 binlog-format 为 ROW 模式，my.cnf 中配置如下

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

docker-compose部署canal

基于已有docker,docker-compose环境；
docker-compose.yml文件如下：

version: '3'
 
services: 
  canal-server: 
    image: canal/canal-server:v1.1.4
    container_name: canal-server
    ports: 
      - 11111:11111
    environment: 
      - canal.instance.mysql.slaveId=12
      - canal.auto.scan=false
      - canal.destinations=nly-v1
      - canal.instance.master.address=192.168.1.129:3306
      - canal.instance.dbUsername=canal
      - canal.instance.dbPassword=canal
      - canal.mq.topic=nly-v1
      - canal.instance.filter.regex=esen_approval.apt_approval
    volumes: 
      - ./canal-server/conf/:/admin/canal-server/conf/
      - ./canal-server/logs/:/admin/canal-server/logs/

canal.instance.mysql.slaveId：slaveId不能与mysql的serverId一样
canal.instance.master.address：mysql地址
canal.instance.dbUsername：mysql账号
canal.instance.dbPassword：mysql密码
更多配置项介绍

启动：docker-compose up -d
新建测试项目添加mvn依赖：

<dependency>
    <groupId>com.alibaba.otter</groupId>
    <artifactId>canal.client</artifactId>
    <version>1.1.4</version>
</dependency>

测试类demo:

import com.alibaba.google.common.collect.Multiset;
import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.client.CanalConnectors;
import com.alibaba.otter.canal.common.utils.AddressUtils;
import com.alibaba.otter.canal.protocol.CanalEntry;
import com.alibaba.otter.canal.protocol.Message;
import com.alibaba.otter.canal.protocol.CanalEntry.Column;
import com.alibaba.otter.canal.protocol.CanalEntry.Entry;
import com.alibaba.otter.canal.protocol.CanalEntry.EntryType;
import com.alibaba.otter.canal.protocol.CanalEntry.RowData;
import com.alibaba.otter.canal.protocol.CanalEntry.RowChange;
 
import java.net.InetSocketAddress;
import java.util.List;
 
/**
 * @ Description：Canal初步测试
 * @ Author     ：wtao
 * @ Date       ：Created in 2020/7/9 13:05
 * @ Modified By：
 * @ Version 1.0.0
 */
public class SimpleCanalClientExample {
    public static void main(String[] args) {
        CanalConnector canalConnector = CanalConnectors.newSingleConnector(new InetSocketAddress("192.168.85.129",
                11111),"good-db","canal","canal");
        int batchSize = 1000;
        int emptyCount = 0;
 
        try{
            canalConnector.connect();
            //监听的表，    格式为：数据库.表名,数据库.表名
            canalConnector.subscribe(".*\\..*");
            canalConnector.rollback();
            int totalEmptyCount  = 120;
            while (emptyCount < totalEmptyCount){
                Message message = canalConnector.getWithoutAck(batchSize);//获取指定数量的数据
                long batchId = message.getId();
                int size = message.getEntries().size();
                if(batchId == -1 || size ==0){
                    emptyCount++;
                    System.out.println("empty count:"+ emptyCount);
                    try{
                        Thread.sleep(1000);
                    }catch (InterruptedException e){}
                }else{
                    emptyCount =0;
                    printEntry(message.getEntries());
                }
                canalConnector.ack(batchId);
            }
            System.out.println("empty too many times, exit");
        }catch (Exception e){
            System.out.println(e);
        }finally {
            canalConnector.disconnect();
 
        }
 
    }
 
    private  static void printEntry(List<Entry> entrys){
        for(Entry entry : entrys){
            if(entry.getEntryType() == EntryType.TRANSACTIONBEGIN || entry.getEntryType() == EntryType.TRANSACTIONEND){
                continue;
            }
            RowChange rowChange = null;
            try {
                rowChange = RowChange.parseFrom(entry.getStoreValue());
 
            }catch (Exception e){}
 
            CanalEntry.EventType eventType = rowChange.getEventType();
            System.out.println(String.format("=======&gt; binlog[%s:%s] , name[%s,%s] ,eventType: %s",
                    entry.getHeader().getLogfileName(),entry.getHeader().getLogfileOffset(),entry.getHeader().getSchemaName(),entry.getHeader().getTableName(),eventType
                    ));
 
            for(RowData rowData:rowChange.getRowDatasList()){
                if(eventType == CanalEntry.EventType.DELETE){
                    printColumn(rowData.getBeforeColumnsList());
                }else if(eventType == CanalEntry.EventType.INSERT){
                    printColumn(rowData.getAfterColumnsList());
                }else{
                    printColumn(rowData.getBeforeColumnsList());
                    printColumn(rowData.getAfterColumnsList());
                }
            }
 
 
        }
 
    }
    private static void printColumn(List<Column> columns){
        for (CanalEntry.Column column:columns
        ) {
            System.out.println(column.getName()+" : "+column.getValue()+"   update="+column.getUpdated() );
        }
    }
}

启动测试类后，执行插入或更新sql,就可以在控制台看到变动日志了