[mysqld]
# 開啟 Binlog 日志
log-bin=mysql-bin
# 設置 Binlog 格式為 ROW 模式
binlog-format=ROW
# 設置唯一的 server-id
server-id=1

# 登錄 MySQL
mysql -u root -p
# 創(chuàng)建用戶
CREATE USER 'canal'@'%' IDENTIFIED BY 'canal';
# 授予權(quán)限
GRANT REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO 'canal'@'%' IDENTIFIED BY 'canal';
# 刷新權(quán)限
FLUSH PRIVILEGES;

ALTER USER 'canal'@'%' IDENTIFIED BY 'canal';

# 主庫的地址和端口
canal.instance.master.address=db_1_ip:3306
# Canal 連接主庫的用戶名和密碼
canal.instance.dbUsername=canal
canal.instance.dbPassword=canal
# 需要同步的表正則表達式，這里表示同步 db_1 數(shù)據(jù)庫的所有表
canal.instance.filter.regex=db_1\\..*

# 主庫的地址和端口
canal.instance.master.address=db_2_ip:3306
# Canal 連接主庫的用戶名和密碼
canal.instance.dbUsername=canal
canal.instance.dbPassword=canal
# 需要同步的表正則表達式，這里表示同步 db_2 數(shù)據(jù)庫的所有表
canal.instance.filter.regex=db_2\\..*

nohup sh bin/canal.sh start &

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

canal:
  server.ip: canal_server_ip
  server.port: 11111

@Service
public class CanalClientService {
    private final CanalConnector canalConnector;
    public CanalClientService(@Value("${canal.server.ip}") String canalServerIp, @Value("${canal.server.port}") int canalServerPort) {
        this.canalConnector = CanalConnectors.newSingleConnector(new InetSocketAddress(canalServerIp, canalServerPort), "example", "", "");
    }
    @PostConstruct
    public void start() {
        canalConnector.connect();
        canalConnector.subscribe("db_1..*, db_2..*"); // 訂閱 db_1 和 db_2 的所有表
        new Thread(this::process).start();
    }
    private void process() {
        while (true) {
            Message message = canalConnector.getWithoutAck(100);
            long batchId = message.getId();
            if (batchId == -1 || message.getEntries().isEmpty()) {
                continue;
            }
            for (Entry entry : message.getEntries()) {
                handleData(entry);
            }
            canalConnector.ack(batchId);
        }
    }
    private void handleData(Entry entry) {
        String schemaName = entry.getHeader().getSchemaName(); // 數(shù)據(jù)庫名
        String tableName = entry.getHeader().getTableName();  // 表名
        EventType eventType = entry.getHeader().getEventType(); // 數(shù)據(jù)變更類型
        System.out.println("Schema: " + schemaName + ", Table: " + tableName + ", Type: " + eventType);
        // 根據(jù)來源數(shù)據(jù)庫同步到對應的從庫
        if ("db_1".equals(schemaName)) {
            syncToBackupDbs(entry, "db_1_bk_1", "db_1_bk_2");
        } else if ("db_2".equals(schemaName)) {
            syncToBackupDbs(entry, "db_2_bk_1", "db_2_bk_2");
        }
    }
    private void syncToBackupDbs(Entry entry, String... backupDbs) {
        // 根據(jù)事件類型同步到從庫
        if (entry.getHeader().getEventType() == EventType.INSERT) {
            for (String db : backupDbs) {
                syncInsert(entry, db);
            }
        } else if (entry.getHeader().getEventType() == EventType.UPDATE) {
            for (String db : backupDbs) {
                syncUpdate(entry, db);
            }
        } else if (entry.getHeader().getEventType() == EventType.DELETE) {
            for (String db : backupDbs) {
                syncDelete(entry, db);
            }
        }
    }
    private void syncInsert(Entry entry, String backupDb) {
        // 使用 MyBatis 將數(shù)據(jù)插入到對應的從庫
        System.out.println("INSERT into " + backupDb);
    }
    private void syncUpdate(Entry entry, String backupDb) {
        // 使用 MyBatis 將數(shù)據(jù)更新到對應的從庫
        System.out.println("UPDATE into " + backupDb);
    }
    private void syncDelete(Entry entry, String backupDb) {
        // 使用 MyBatis 將數(shù)據(jù)從對應的從庫刪除
        System.out.println("DELETE from " + backupDb);
    }
}