Canal实现MYSQL实时数据同步

Canal 实现 MySQL 实时数据同步的基本步骤如下：

1. 部署 Canal 服务器。
2. 配置 MySQL 以兼容模式运行，并为 Canal 创建相应的复制账号。
3. 启动 Canal 服务器，并连接到 MySQL 数据库。
4. 配置 Canal 实例，指定需要同步的数据库或表。
5. 启动 Canal 实例，开始监听并同步 MySQL 的数据变化。
6. 数据变化会以特定的格式发送到 Canal 客户端。

以下是一个简化的示例，展示如何使用 Java 客户端接收 Canal 服务器的数据变化：

import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.client.CanalConnectors;
import com.alibaba.otter.canal.protocol.Message;
import com.alibaba.otter.canal.protocol.CanalEntry;
 
public class SimpleCanalClientExample {
 
    public static void main(String args[]) {
        // 创建连接
        CanalConnector connector = CanalConnectors.newSingleConnector(
                new InetSocketAddress(AddressUtils.getHostIp(),
                11111), "example", "", "");
 
        int batchSize = 1000;
        try {
            connector.connect();
            connector.subscribe(".*\\..*");
            connector.rollback();
            while (true) {
                Message message = connector.getWithoutAck(batchSize); // 获取指定数量的数据
                long batchId = message.getId();
                if (batchId == -1 || message.getEntries().isEmpty()) {
                    // 没有数据，休眠一会儿
                    Thread.sleep(1000);
                } else {
                    dataHandle(message.getEntries());
                    connector.ack(batchId); // 确认消息消费成功
                }
            }
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            connector.disconnect();
        }
    }
 
    private static void dataHandle(List<CanalEntry.Entry> entries) {
        if (entries == null || entries.isEmpty()) {
            return;
        }
        for (CanalEntry.Entry entry : entries) {
            if (entry.getEntryType() == CanalEntry.EntryType.TRANSACTIONBEGIN || entry.getEntryType() == CanalEntry.EntryType.TRANSACTIONEND) {
                // 事务开始/结束
            } else if (entry.getEntryType() == CanalEntry.EntryType.ROWDATA) {
                CanalEntry.RowChange rowChage = CanalEntry.RowChange.parseFrom(entry.getStoreValue());
                // 处理具体的数据
            }
        }
    }
}

在这个示例中，我们创建了一个连接到 Canal 服务器的客户端，订阅了所有的数据库和表。然后，我们进入一个循环，不断地获取数据变化并处理它们。数据变化类型有事务开始/结束和行数据变化，我们需要对这些类型进行区分并相应地处理。