使用TCP方式拉取Canal数据

矫情吗；* 2024-03-03 08:08 78阅读 0赞

### 1 Canal对接Kafka联调 ###

#### 1.1 配置修改 ####

##### canal.properties #####

![3650dc4bb8544fa5b68a6979cd3a99bd.png][]

修改 zk：

canal.zkServers = 10.51.50.219:2181

##### instance.properties #####

开启配置项：

`canal.mq.dynamicTopic` 是 Canal 的 MQ 动态 Topic 配置项：

*  `test_javaedge_01` 是kafka 的 topic
 *  `test_db.users` 要监控的数据库、表
 *  当 `test_db.users` 表发生变化时，Canal 将会把变化的数据推送到名为 `test_javaedge_01:test_db.users` 的 MQ Topic 中。

canal.mq.dynamicTopic=test_javaedge_01:test_db\\.users

开启一个消费者

[root@javaedge-kafka-dev bin]# bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test_javaedge_01

datagrip 新增数据：

![a5c329df327d418f82ad5dde1e8cea4d.png][]

消费到该数据：

![263f34fbefbe40789f90b5e5b44198ff.png][]

### 2 使用TCP方式拉取Canal数据 ###

现在 serverMode 改回tcp。重启

javaedge@JavaEdgedeMac-mini deployer % jps
    71002 CanalLauncher
    javaedge@JavaEdgedeMac-mini deployer %

#### canal 同步程序 ####

package com.javaedge.canal;
    
    import com.alibaba.fastjson.JSON;
    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.common.base.CaseFormat;
    
    import java.net.InetSocketAddress;
    import java.util.HashMap;
    import java.util.List;
    
    public class CanalClientApp {
        
        public static void main(String[] args) throws Exception {
        
    
            CanalConnector connector = CanalConnectors.newSingleConnector(
                    new InetSocketAddress("localhost", 11111),
                    "example",
                    null, null);
    
            while (true) {
        
                connector.connect();
                connector.subscribe("test_db.users");
                Message message = connector.get(100);
                List<CanalEntry.Entry> entries = message.getEntries();
                if (entries.size()>0) {
        
                    for (CanalEntry.Entry entry : entries) {
        
                        String tableName = entry.getHeader().getTableName();
    
                        CanalEntry.RowChange rowChange = CanalEntry.RowChange.parseFrom(entry.getStoreValue());
                        List<CanalEntry.RowData> rowDatasList = rowChange.getRowDatasList();
                        CanalEntry.EventType eventType = rowChange.getEventType();
    
                        if (eventType == CanalEntry.EventType.INSERT) {
        
                            for (CanalEntry.RowData rowData : rowDatasList) {
        
                                List<CanalEntry.Column> afterColumnsList = rowData.getAfterColumnsList();
                                HashMap<Object, Object> map = new HashMap<>();
                                for (CanalEntry.Column column : afterColumnsList) {
        
                                    String key = CaseFormat.LOWER_UNDERSCORE.to(CaseFormat.LOWER_CAMEL, column.getName());
                                    map.put(key, column.getValue());
                                }
                                System.out.println("tableName=" + tableName + "  map=" + JSON.toJSONString(map));
                            }
                        }
                    }
    
                }
    
            }
    
        }
    }

运行程序。操作 user 数据表，新增一行数据：

![cd8e9d994a774fa7a7f079be05dd981a.png][]

程序输出：

![c7cff492be15459a92773d6c3884c698.png][]

显然，后续不管你想把数据同步到哪儿去，都完全自由！

#### 数据链路 ####

MySQL -》canal server（tcp）-》canal client-》kafka。

[3650dc4bb8544fa5b68a6979cd3a99bd.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/03/8d71fa7f633f4151899bddec5aab6e92.png
[a5c329df327d418f82ad5dde1e8cea4d.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/03/2b5fcb9cd8954c08b56c297acc5d77f2.png
[263f34fbefbe40789f90b5e5b44198ff.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/03/c5143542d0714ee6bc6c11878bf0b66d.png
[cd8e9d994a774fa7a7f079be05dd981a.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/03/5b58ef5b0d3c45029a6d99d2726e739a.png
[c7cff492be15459a92773d6c3884c698.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/03/cae85e464b234a588048b8d6e2cd62f9.png