导包

<!-- https://mvnrepository.com/artifact/com.alibaba.otter/canal.client -->
<dependency>
  <groupId>com.alibaba.otter</groupId>
  <artifactId>canal.client</artifactId>
  <version>1.1.0</version>
</dependency>
<!-- https://mvnrepository.com/artifact/org.apache.kafka/kafka-clients -->
<dependency>
  <groupId>org.apache.kafka</groupId>
  <artifactId>kafka-clients</artifactId>
  <version>3.1.0</version>
</dependency>

例子


import com.alibaba.fastjson.JSONObject;
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.protobuf.ByteString;
import com.google.protobuf.InvalidProtocolBufferException;
import java.net.InetSocketAddress;
import java.util.List;
public class CanalClient {
    public static void main(String[] args) {
        // 获取连接
        // 这里填写canal所配置的服务器ip,端口号,destination(在canal.properties文件里)以及服务器账号密码
        CanalConnector canalConnector = CanalConnectors.newSingleConnector(
                new InetSocketAddress("127.0.0.1", 11111),
                "example", "", "");
        try {
            // 打开连接
            canalConnector.connect();
            // 订阅数据库 gamll 库的 所有表
            canalConnector.subscribe("gmall\\..*");// .*代表database，..*代表table
            //订阅数据库表,全部表
//            canalConnector.subscribe(".*\\..*");
            //回滚到未进行ack的地方，下次fetch的时候，可以从最后一个没有ack的地方开始拿
            canalConnector.rollback();
            while (true) {
                // 获取指定数量 数据
                Message message = canalConnector.getWithoutAck(100);
//            Message message = canalConnector.get(100);
                //获取批量ID
                long batchId = message.getId();
                // 获取Entry集合 数量
                List<CanalEntry.Entry> entries = message.getEntries();
                // 判断集合是否为空,如果为空,则等待一会继续拉取数据
                if (batchId == -1 || entries.size() <= 0) {
                    System.out.println("当前抓取没有数据，休息一会。。。。。。");
                    Thread.sleep(10000);
                } else {
                    // 遍历entries，单条解析
                    printEntry(entries);
                }
                //进行 batch id 的确认。确认之后，小于等于此 batchId 的 Message 都会被确认。
                canalConnector.ack(batchId);
//                canalConnector.rollback(batchId); // 处理失败, 回滚数据
            }
        } catch (Throwable e) {
            e.printStackTrace();
        } finally {
            canalConnector.disconnect();
        }
    }
    /**
     * 打印canal server解析binlog获得的实体类信息
     */
    private static void printEntry(List<CanalEntry.Entry> entries) throws InvalidProtocolBufferException {
        for (CanalEntry.Entry entry : entries) {
            //1.获取表名
            String tableName = entry.getHeader().getTableName();
            //2.获取类型
            CanalEntry.EntryType entryType = entry.getEntryType();
            if (entry.getEntryType() == CanalEntry.EntryType.TRANSACTIONBEGIN || entry.getEntryType() == CanalEntry.EntryType.TRANSACTIONEND) {
                //开启/关闭事务的实体类型，跳过
                continue;
            }
            //3.获取序列化后的数据
            ByteString storeValue = entry.getStoreValue();
            //4.判断当前entryType类型是否为ROWDATA
            if (CanalEntry.EntryType.ROWDATA.equals(entryType)) {
                //5.反序列化数据
                //RowChange对象，包含了一行数据变化的所有特征
                //比如isDdl 是否是ddl变更操作 sql 具体的ddl sql beforeColumns afterColumns 变更前后的数据字段等等
                CanalEntry.RowChange rowChange;
                try {
                    rowChange = CanalEntry.RowChange.parseFrom(entry.getStoreValue());
                } catch (Throwable e) {
                    throw new RuntimeException("ERROR ## canal解析出错, data:" + entry.toString(), e);
                }
                //6.获取当前事件的操作类型, insert/update/delete类型
                CanalEntry.EventType eventType = rowChange.getEventType();
                //打印Header信息
                System.out.println(String.format("================》; binlog[%s:%s] , name[%s,%s] , eventType : %s",
                        entry.getHeader().getLogfileName(), entry.getHeader().getLogfileOffset(),
                        entry.getHeader().getSchemaName(), entry.getHeader().getTableName(),
                        eventType));
                //判断是否是DDL语句
                if (rowChange.getIsDdl()) {
                    System.out.println("================》;isDDl: true,sql:" + rowChange.getSql());
                }
                //7.获取数据集
                List<CanalEntry.RowData> rowDataList = rowChange.getRowDatasList();
                //8.遍历rowDataList，并打印数据集
                for (CanalEntry.RowData rowData : rowDataList) {
                    // 判断类型
                    // eventType == CanalEntry.EventType.DELETE
                    //如果是删除语句
//                    if (eventType == CanalEntry.EventType.DELETE) {
//                        printColumn(rowData.getBeforeColumnsList());
//                        //如果是新增语句
//                    } else if (eventType == CanalEntry.EventType.INSERT) {
//                        printColumn(rowData.getAfterColumnsList());
//                        //如果是更新的语句
//                    } else {
//                        //变更前的数据
//                        System.out.println("------->; before");
//                        printColumn(rowData.getBeforeColumnsList());
//                        //变更后的数据
//                        System.out.println("------->; after");
//                        printColumn(rowData.getAfterColumnsList());
//                    }
                    JSONObject beforeData = new JSONObject();
                    List<CanalEntry.Column> beforeColumnsList = rowData.getBeforeColumnsList();
                    for (CanalEntry.Column column : beforeColumnsList) {
                        beforeData.put(column.getName(), column.getValue());
                    }
                    JSONObject afterData = new JSONObject();
                    List<CanalEntry.Column> afterColumnsList = rowData.getAfterColumnsList();
                    for (CanalEntry.Column column : afterColumnsList) {
                        afterData.put(column.getName(), column.getValue());
                    }
                    //数据打印
                    System.out.println("Table:" + tableName + System.lineSeparator() +
                            "EventType:" + eventType + System.lineSeparator() +
                            "之前数据是:" + beforeData + System.lineSeparator() +
                            "之后数据是:" + afterData);
                }
            } else {
                System.out.println("当前操作类型为：" + entryType);
            }
        }
    }
    private static void printColumn(List<CanalEntry.Column> columns) {
        for (CanalEntry.Column column : columns) {
            System.out.println(column.getName() + " : " + column.getValue() + "    update=" + column.getUpdated());
        }
    }
}

输出

================》; binlog[mysql-bin.000004:3069] , name[gmall,user_info] , eventType : UPDATE
Table:user_info
EventType:UPDATE
之前数据是:{"sex":"0","name":"a22","id":"4"}
之后数据是:{"sex":"1","name":"a22","id":"4"}

当前抓取没有数据，休息一会。。。。。。

================》; binlog[mysql-bin.000004:3227] , name[gmall,user_info_3] , eventType : CREATE
================》;isDDl: true,sql:CREATE TABLE `user_info_3` (
  `id` int(11) NOT NULL AUTO_INCREMENT,
  `name` varchar(255) COLLATE utf8mb4_unicode_ci DEFAULT NULL,
  `sex` varchar(255) COLLATE utf8mb4_unicode_ci DEFAULT NULL,
  PRIMARY KEY (`id`)
) ENGINE=InnoDB AUTO_INCREMENT=5 DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci

tcp测试

导包

例子