登錄mysql
mysql -h 127.0.0.1 -P 3306 -u root -p
創(chuàng)建同步用戶 repl 密碼設為123456
CREATE USER 'repl'@'%' IDENTIFIED BY '123456';
給予同步權限
GRANT REPLICATION SLAVE ON *.* to 'repl'@'%' identified by '123456';
給予repl只讀test庫的權限，test庫是用來同步數據的
GRANT SELECT ON test.* to 'repl'@'%' identified by '123456';
canal_manager是canal admin需要的，給予repl對該庫的讀寫權限
GRANT ALL PRIVILEGES ON canal_manager.* to 'repl'@'%' identified by '123456';
mysql my.cnf配置文件增加主從配置master數據庫的配置信息
#主數據主從配置 唯一id
server_id=1
#開啟logbin
log-bin=mysql-bin
#寫入模式 row
binlog-format=ROW
#需要同步的庫
binlog-do-db=test
#忽略的數據庫
replicate-ignore-db=mysql
replicate-ignore-db=sys
replicate-ignore-db=information_schema
replicate-ignore-db=performance_schema

server:
  port: 8089
spring:
  jackson:
    date-format: yyyy-MM-dd HH:mm:ss
    time-zone: GMT+8
spring.datasource:
  address: 127.0.0.1:3306
  database: canal_manager
  username: repl
  password: 123456
  driver-class-name: com.mysql.jdbc.Driver
  url: jdbc:mysql://${spring.datasource.address}/${spring.datasource.database}?useUnicode=true&characterEncoding=UTF-8&useSSL=false&allowPublicKeyRetrieval=true
  hikari:
    maximum-pool-size: 30
    minimum-idle: 1
canal:
  adminUser: admin
  adminPasswd: 123456

./startup.sh

mv canal.properties canal.properties_bak
cp canal_local.properties canal.properties
vim canal.properties

# register ip
canal.register.ip =
# canal admin config  canalAdmin 的鏈接、端口、用戶名和MD5密碼
canal.admin.manager = 127.0.0.1:8089
canal.admin.port = 11110
canal.admin.user = admin
canal.admin.passwd =6F32482BAFC60F23B7736044CEFC1799166E5CDB
# admin auto register canal server啟動后自動注入到canal admin管理模塊
canal.admin.register.auto = true
canal.admin.register.cluster =
canal.admin.register.name =

./startup.sh

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

package com.hy.das.config;
import com.alibaba.fastjson.JSONObject;
import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.client.CanalConnectors;
import com.alibaba.otter.canal.protocol.CanalEntry;
import com.alibaba.otter.canal.protocol.Message;
import com.google.protobuf.ByteString;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.stereotype.Component;
import java.net.InetSocketAddress;
import java.util.List;
@Component
public class CanalClient implements InitializingBean{
    private final static int BATCH_SIZE = 1000;
    @Override
    public void afterPropertiesSet() throws Exception {
        // 創(chuàng)建鏈接 此處的11111為tcp端口 在canal admin Server管理模塊可以查看
        CanalConnector connector = CanalConnectors.newSingleConnector(new InetSocketAddress("127.0.0.1", 11111),
                "test", "", "");
        try {
            //打開連接
            connector.connect();
            //訂閱數據庫表,全部表
            connector.subscribe(".*\\..*");
            //回滾到未進行ack的地方，下次fetch的時候，可以從最后一個沒有ack的地方開始拿
            connector.rollback();
            while (true) {
                // 獲取指定數量的數據
                Message message = connector.getWithoutAck(BATCH_SIZE);
                System.out.println(message.getEntries().size());
                //獲取批量ID
                long batchId = message.getId();
                //獲取批量的數量
                int size = message.getEntries().size();
                //如果沒有數據
                if (batchId == -1 || size == 0) {
                    try {
                        //線程休眠2秒
                        Thread.sleep(2000);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                } else {
                    System.out.println("----------------");
                    //如果有數據,處理數據
                    //遍歷entries，單條解析
                    for (CanalEntry.Entry entry : message.getEntries()) {
                        //獲取表名
                        String tableName = entry.getHeader().getTableName();
                        //獲取類型
                        CanalEntry.EntryType entryType = entry.getEntryType();
                        //獲取序列化后的數據
                        ByteString storeValue = entry.getStoreValue();
                        //判斷entry類型是否為ROWDATA類型
                        if (CanalEntry.EntryType.ROWDATA.equals(entryType)){
                            //反序列化
                            CanalEntry.RowChange rowChange = CanalEntry.RowChange.parseFrom(storeValue);
                            //獲取當前事件操作類型
                            CanalEntry.EventType eventType = rowChange.getEventType();
                            //獲取數據集
                            List<CanalEntry.RowData> rowDatasList = rowChange.getRowDatasList();
                            //遍歷
                            for (CanalEntry.RowData rowData : rowDatasList) {
                                //改變前數據
                                JSONObject jsonObjectBefore = new JSONObject();
                                List<CanalEntry.Column> beforeColumnsList = rowData.getBeforeColumnsList();
                                for (CanalEntry.Column column : beforeColumnsList) {
                                    jsonObjectBefore.put(column.getName(),column.getValue());
                                }
                                //改變后數據
                                JSONObject jsonObjectAfter = new JSONObject();
                                List<CanalEntry.Column> afterColumnsList = rowData.getAfterColumnsList();
                                for (CanalEntry.Column column : afterColumnsList) {
                                    jsonObjectAfter.put(column.getName(),column.getValue());
                                }
                                System.out.println("Table:"+tableName+",EventTpye:"+eventType+",Before:"+jsonObjectBefore+",After:"+jsonObjectAfter);
                            }
                        }else {
                            System.out.println("當前操作類型為："+entryType);
                        }
                    }
                }
                //進行 batch id 的確認。確認之后，小于等于此 batchId 的 Message 都會被確認。
                connector.ack(batchId);
            }
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            connector.disconnect();
        }
    }
}

#數據變更發(fā)送到kafka
# 設置輸出目標為 kafka
canal.serverMode = kafka
# Kafka 地址
canal.mq.servers  = xx.xx.xx.xx:9092
# 投遞失敗的重試次數，默認0，改為2
canal.mq.retries = 2
# Kafka batch.size，即producer一個微批次的大小，默認16K，這里加倍
canal.mq.batchSize = 32768
# Kafka max.request.size，即一個請求的最大大小，默認1M，這里也加倍
canal.mq.maxRequestSize = 2097152
# Kafka linger.ms，即sender線程在檢查微批次是否就緒時的超時，默認0ms，這里改為200ms
# 滿足batch.size和linger.ms其中之一，就會發(fā)送消息
canal.mq.lingerMs = 200
# Kafka buffer.memory，緩存大小，默認32M
canal.mq.bufferMemory = 33554432
# 獲取binlog數據的批次大小，默認50
canal.mq.canalBatchSize = 50
# 獲取binlog數據的超時時間，默認200ms
canal.mq.canalGetTimeout = 200
# 是否將binlog轉為JSON格式。如果為false，就是原生Protobuf格式
canal.mq.flatMessage = true
# 壓縮類型，官方文檔沒有描述
canal.mq.compressionType = none
# Kafka acks，默認all，表示分區(qū)leader會等所有follower同步完才給producer發(fā)送ack
# 0表示不等待ack，1表示leader寫入完畢之后直接ack
canal.mq.acks = all
# Kafka消息投遞是否使用事務
# 主要針對flatMessage的異步發(fā)送和動態(tài)多topic消息投遞進行事務控制來保持和Canal binlog位置的一致性
# flatMessage模式下建議開啟
canal.mq.transaction = true