java canal例子

Canal是阿里巴巴开源的一款基于MySQL数据库的数据增量订阅和消费组件，可以将MySQL中的数据变更实时推送到消息队列中，实现数据的异步传输。以下是一个简单的Java Canal示例：

1. 引入Canal客户端依赖：

<dependency>
    <groupId>com.alibaba.otter</groupId>
    <artifactId>canal.client</artifactId>
    <version>1.1.4</version>
</dependency>

2. 编写Canal客户端代码：

public class CanalClientExample {

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

        try {
            // 连接Canal服务器
            connector.connect();
            // 订阅指定的数据库和表
            connector.subscribe("test.user");
            // 回滚到最后一次的getBatch操作的位置
            connector.rollback();
            while (true) {
                // 获取指定数量的数据变更事件
                Message message = connector.getWithoutAck(1000);
                long batchId = message.getId();
                int size = message.getEntries().size();
                if (batchId == -1 || size == 0) {
                    Thread.sleep(1000);
                } else {
                    // 处理数据变更事件
                    for (CanalEntry.Entry entry : message.getEntries()) {
                        if (entry.getEntryType() == CanalEntry.EntryType.ROWDATA) {
                            RowChange rowChange = null;
                            try {
                                rowChange = RowChange.parseFrom(entry.getStoreValue());
                            } catch (Exception e) {
                                throw new RuntimeException("ERROR ## parser of eromanga-event has an error , data:" + entry.toString(),e);
                            }
                            for (CanalEntry.RowData rowData : rowChange.getRowDatasList()) {
                                if (rowChange.getEventType() == CanalEntry.EventType.DELETE) {
                                    // 删除操作
                                    System.out.println("delete:");
                                    printColumn(rowData.getBeforeColumnsList());
                                } else if (rowChange.getEventType() == CanalEntry.EventType.INSERT) {
                                    // 插入操作
                                    System.out.println("insert:");
                                    printColumn(rowData.getAfterColumnsList());
                                } else {
                                    // 更新操作
                                    System.out.println("update:");
                                    printColumn(rowData.getAfterColumnsList());
                                }
                            }
                        }
                    }
                    // 提交确认
                    connector.ack(batchId);
                }
            }
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            // 关闭Canal连接器
            connector.disconnect();
        }
    }

    private static void printColumn(List<CanalEntry.Column> columns) {
        for (CanalEntry.Column column : columns) {
            System.out.println(column.getName() + " : " + column.getValue() + "  update=" + column.getUpdated());
        }
    }
}

在上面的代码中，我们创建了一个Canal连接器，并订阅了test库中的user表。然后通过循环调用`getWithoutAck()`方法获取指定数量的数据变更事件，对这些事件进行处理后，调用`ack()`方法确认处理成功。最后，在异常情况下，关闭Canal连接器。

需要注意的是，由于Canal是基于binlog实现的，所以在使用Canal时需要确保MySQL的binlog开启，否则会出现无法获取数据变更事件的情况。此外，Canal还支持Kafka、RocketMQ等多种消息队列，可以根据需要选择不同的使用方式。