MySQL数据实时同步Redis的方案全解析

CREATE USER 'canal'@'%' IDENTIFIED BY 'your_password'; GRANT SELECT, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO 'canal'@'%'; FLUSH PRIVILEGES;

 Canal服务端口
canal.port = 11111
# 服务模式，如tcp, kafka, RocketMQ等[citation:6]
canal.serverMode = tcp
# 目的地集合，对应instance配置的文件夹名
canal.destinations = example

 主库地址
canal.instance.master.address = 127.0.0.1:3306
# 数据库用户名
canal.instance.dbUsername = canal
# 数据库密码
canal.instance.dbPassword = your_canal_password
# 要监听的表，支持正则表达式
canal.instance.filter.regex = .*\\..*

// Canal客户端监听Binlog（伪代码）
CanalConnector connector = CanalConnectors.newClusterConnector("127.0.0.1:2181", "example", "", "");
connector.connect();
connector.subscribe(".*\\..*");
while (true) {
    Message message = connector.getWithoutAck(100);
    for (CanalEntry.Entry entry : message.getEntries()) {
        if (entry.getEntryType() == CanalEntry.EntryType.ROWDATA) {
            CanalEntry.RowChange rowChange = CanalEntry.RowChange.parseFrom(entry.getStoreValue());
            for (CanalEntry.RowData rowData : rowChange.getRowDatasList()) {
                String tableName = entry.getHeader().getTableName();
                // 根据表名和主键构造Redis Key，这里假设第一列为主键
                String key = "cache:" + tableName + ":" + rowData.getBeforeColumns(0).getValue();
                if (rowChange.getEventType() == CanalEntry.EventType.DELETE) {
                    redis.del(key);  // 删除缓存[citation:6]
                } else {
                    // 更新缓存：这里可以将rowData的AfterColumns序列化后存储
                    redis.set(key, serialize(rowData.getAfterColumnsList()));  // 更新缓存[citation:6]
                }
            }
        }
    }
}

d /your/target/directory/bin
sh startup.sh