浅谈canal实例 在docker里面安装canal镜像 Canal监听MySQL数据库变更并同步更新Redis和Elasticsearch 示例

目录

1. 环境准备

1.1 MySQL配置

1.2 部署Canal Server

2. Spring Boot项目配置

2.1 添加依赖

2.2 配置参数

3. 实现Canal监听与同步

3.1 Canal客户端监听

3.2 同步到Redis

3.3 同步到Elasticsearch

4. 注意事项

在Spring Boot中通过Canal监听MySQL数据库变更并同步更新Redis和Elasticsearch，可按照以下步骤实现：

1. 环境准备

1.1 MySQL配置

• 开启Binlog并设置为ROW模式：

[mysqld]
log-bin=mysql-bin
binlog-format=ROW
server-id=1

• 创建Canal用户并授权：

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

1.2 部署Canal Server

1. 下载Canal Server：Canal Releases
2. 修改配置 conf/example/instance.properties ：

canal.instance.master.address=127.0.0.1:3306 
canal.instance.dbUsername=canal 
canal.instance.dbPassword=canal 
canal.instance.filter.regex=.*\\..*  # 监听所有库表，或指定如test.user

2. Spring Boot项目配置

2.1 添加依赖

<!-- Canal客户端 -->
<dependency>
    <groupId>com.alibaba.otter</groupId> 
    <artifactId>canal.client</artifactId> 
    <version>1.1.6</version>
</dependency>
<!-- Redis -->
<dependency>
    <groupId>org.springframework.boot</groupId> 
    <artifactId>spring-boot-starter-data-redis</artifactId>
</dependency>
<!-- Elasticsearch -->
<dependency>
    <groupId>org.springframework.boot</groupId> 
    <artifactId>spring-boot-starter-data-elasticsearch</artifactId>
</dependency>

2.2 配置参数

application.yml ：

canal:
  server: 127.0.0.1:11111
  destination: example
  username: canal
  password: canal

spring:
  redis:
    host: localhost
    port: 6379
  data:
    elasticsearch:
      cluster-nodes: localhost:9200

3. 实现Canal监听与同步

3.1 Canal客户端监听

@Component
public class CanalListener {

    @Autowired
    private RedisTemplate<String, Object> redisTemplate;
    @Autowired
    private ElasticsearchRestTemplate esTemplate;

    @PostConstruct
    public void init() {
        CanalConnector connector = CanalConnectors.newSingleConnector( 
                new InetSocketAddress("127.0.0.1", 11111), "example", "canal", "canal");
        
        Thread thread = new Thread(() -> {
            connector.connect(); 
            connector.subscribe(".*\\..*"); 
            while (true) {
                Message message = connector.getWithoutAck(100); 
                long batchId = message.getId(); 
                if (batchId != -1) {
                    processEntry(message.getEntries()); 
                    connector.ack(batchId); 
                }
            }
        });
        thread.start(); 
    }

    private void processEntry(List<Entry> entries) {
        for (Entry entry : entries) {
            if (entry.getEntryType()  == EntryType.ROWDATA) {
                RowChange rowChange = RowChange.parseFrom(entry.getStoreValue()); 
                for (RowData rowData : rowChange.getRowDatasList())  {
                    String tableName = entry.getHeader().getTableName(); 
                    EventType eventType = rowChange.getEventType(); 
                    
                    // 解析变更前后的数据
                    Map<String, String> before = parseColumns(rowData.getBeforeColumnsList()); 
                    Map<String, String> after = parseColumns(rowData.getAfterColumnsList()); 
                    
                    // 根据事件类型同步数据
                    switch (eventType) {
                        case INSERT:
                        case UPDATE:
                            syncToRedis(tableName, after);
                            syncToElasticsearch(tableName, after);
                            break;
                        case DELETE:
                            deleteFromRedis(tableName, before);
                            deleteFromElasticsearch(tableName, before);
                            break;
                    }
                }
            }
        }
    }

    private Map<String, String> parseColumns(List<Column> columns) {
        return columns.stream() 
                .collect(Collectors.toMap(Column::getName,  Column::getValue));
    }
}

3.2 同步到Redis

private void syncToRedis(String tableName, Map<String, String> data) {
    String key = tableName + ":" + data.get("id");  // 假设主键为id
    redisTemplate.opsForValue().set(key,  data);
}

private void deleteFromRedis(String tableName, Map<String, String> data) {
    String key = tableName + ":" + data.get("id"); 
    redisTemplate.delete(key); 
}

3.3 同步到Elasticsearch

private void syncToElasticsearch(String tableName, Map<String, String> data) {
    IndexQuery indexQuery = new IndexQueryBuilder()
            .withId(data.get("id")) 
            .withObject(data)
            .build();
    esTemplate.index(indexQuery,  IndexCoordinates.of(tableName)); 
}

private void deleteFromElasticsearch(String tableName, Map<String, String> data) {
    esTemplate.delete(data.get("id"),  IndexCoordinates.of(tableName)); 
}

4. 注意事项

1. 异常处理：增加重试机制或记录错误日志，确保网络波动时的数据一致性。
2. 性能优化：批量处理Canal消息，减少Redis/ES的频繁写入。
3. 数据结构：确保Elasticsearch的索引Mapping与MySQL表结构兼容。
4. 事务管理：如需强一致性，可结合本地事务表或消息队列（如RocketMQ）做可靠投递。

通过以上步骤，Spring Boot应用能够实时监听MySQL变更，并自动同步到Redis和Elasticsearch，保障数据一致性。