canal客户端连接canal集群demo

下面是一个使用 Canal 客户端连接 Canal 集群的简单示例代码，可以供参考：

import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.client.CanalConnectors;
import com.alibaba.otter.canal.common.utils.AddressUtils;
import com.alibaba.otter.canal.protocol.CanalEntry;
import com.alibaba.otter.canal.protocol.Message;

import java.net.InetSocketAddress;
import java.util.List;

public class CanalClientDemo {

    public static void main(String[] args) {
        // 创建连接器对象
        CanalConnector connector = CanalConnectors.newSingleConnector(
                new InetSocketAddress(AddressUtils.getHostIp(), 11111),
                "example", "", "");

        try {
            // 连接 Canal 服务器
            connector.connect();
            connector.subscribe(".*\\..*");
            connector.rollback();

            // 循环读取 Canal 中的数据变更
            while (true) {
                Message message = connector.getWithoutAck(100);
                long batchId = message.getId();
                int size = message.getEntries().size();
                if (batchId == -1 || size == 0) {
                    Thread.sleep(1000);
                } else {
                    List<CanalEntry.Entry> entries = message.getEntries();
                    for (CanalEntry.Entry entry : entries) {
                        if (entry.getEntryType() == CanalEntry.EntryType.ROWDATA) {
                            CanalEntry.RowChange rowChange = CanalEntry.RowChange.parseFrom(entry.getStoreValue());
                            String schemaName = entry.getHeader().getSchemaName();
                            String tableName = entry.getHeader().getTableName();
                            String eventType = rowChange.getEventType().toString();
                            for (CanalEntry.RowData rowData : rowChange.getRowDatasList()) {
                                // 处理数据变更
                                System.out.println(schemaName + "." + tableName + ": " + eventType + " " + rowData);
                            }
                        }
                    }
                    connector.ack(batchId);
                }
            }
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            connector.disconnect();
        }
    }
}

以上代码实现了一个简单的 Canal 客户端，连接到 Canal 服务器后，订阅所有的数据库和表，并循环读取 Canal 中的数据变更。对于每个数据变更，打印出相关的信息，并调用 ack() 方法确认消息已经处理完毕。