1、拉取
    CanalConnector.getWithoutAck(batchSize);

    1. @Component
    2. public class CannalClient implements InitializingBean {
    4.     private final static int BATCH_SIZE = 1000;
    6.     @Override
    7.     public void afterPropertiesSet() throws Exception {
    8.         // 创建链接
    9.         CanalConnector connector = CanalConnectors.newSingleConnector(new InetSocketAddress("127.0.0.1", 11111), "example", "", "");
    10.         try {
    11.             //打开连接
    12.             connector.connect();
    13.             //订阅数据库表,全部表
    14.             connector.subscribe(".*\\..*");
    15.             //回滚到未进行ack的地方，下次fetch的时候，可以从最后一个没有ack的地方开始拿
    16.             connector.rollback();
    17.             while (true) {
    18.                  //尝试从master那边拉去数据batchSize条记录，有多少取多少
    19.                 Message message = connector.getWithoutAck(BATCH_SIZE);
    20.                 //获取批量ID
    21.                 long batchId = message.getId();
    22.                 //获取批量的数量
    23.                 int size = message.getEntries().size();
    24.                 //如果没有数据
    25.                 if (batchId == -1 || size == 0) {
    26.                     try {
    27.                         //线程休眠2秒
    28.                         Thread.sleep(2000);
    29.                     } catch (InterruptedException e) {
    30.                         e.printStackTrace();
    31.                     }
    32.                 } else {
    33.                     //如果有数据,处理数据
    34.                     printEntry(message.getEntries());
    35.                 }
    36.                 //进行 batch id 的确认。确认之后，小于等于此 batchId 的 Message 都会被确认。
    37.                 connector.ack(batchId);
    38.             }
    39.         } catch (Exception e) {
    40.             e.printStackTrace();
    41.         } finally {
    42.             connector.disconnect();
    43.         }
    44.     }
    46.     /**
    47.      * 打印canal server解析binlog获得的实体类信息
    48.      */
    49.     private static void printEntry(List<Entry> entrys) {
    50.         for (Entry entry : entrys) {
    51.             if (entry.getEntryType() == EntryType.TRANSACTIONBEGIN || entry.getEntryType() == EntryType.TRANSACTIONEND) {
    52.                 //开启/关闭事务的实体类型，跳过
    53.                 continue;
    54.             }
    55.             //RowChange对象，包含了一行数据变化的所有特征
    56.             //比如isDdl 是否是ddl变更操作 sql 具体的ddl sql beforeColumns afterColumns 变更前后的数据字段等等
    57.             RowChange rowChage;
    58.             try {
    59.                 rowChage = RowChange.parseFrom(entry.getStoreValue());
    60.             } catch (Exception e) {
    61.                 throw new RuntimeException("ERROR ## parser of eromanga-event has an error , data:" + entry.toString(), e);
    62.             }
    63.             //获取操作类型：insert/update/delete类型
    64.             EventType eventType = rowChage.getEventType();
    65.             //打印Header信息
    66.             System.out.println(String.format("================》; binlog[%s:%s] , name[%s,%s] , eventType : %s",
    67.                     entry.getHeader().getLogfileName(), entry.getHeader().getLogfileOffset(),
    68.                     entry.getHeader().getSchemaName(), entry.getHeader().getTableName(),
    69.                     eventType));
    70.             //判断是否是DDL语句
    71.             if (rowChage.getIsDdl()) {
    72.                 System.out.println("================》;isDdl: true,sql:" + rowChage.getSql());
    73.             }
    74.             //获取RowChange对象里的每一行数据，打印出来
    75.             for (RowData rowData : rowChage.getRowDatasList()) {
    76.                 //如果是删除语句
    77.                 if (eventType == EventType.DELETE) {
    78.                     printColumn(rowData.getBeforeColumnsList());
    79.                     //如果是新增语句
    80.                 } else if (eventType == EventType.INSERT) {
    81.                     printColumn(rowData.getAfterColumnsList());
    82.                     //如果是更新的语句
    83.                 } else {
    84.                     //变更前的数据
    85.                     System.out.println("------->; before");
    86.                     printColumn(rowData.getBeforeColumnsList());
    87.                     //变更后的数据
    88.                     System.out.println("------->; after");
    89.                     printColumn(rowData.getAfterColumnsList());
    90.                 }
    91.             }
    92.         }
    93.     }
    95.     private static void printColumn(List<Column> columns) {
    96.         for (Column column : columns) {
    97.             System.out.println(column.getName() + " : " + column.getValue() + "    update=" + column.getUpdated());
    98.         }
    99.     }
    100. }

    2、监听

    
@CanalEventListener
public class MyEventListener {

    @InsertListenPoint
    public void onEvent(CanalEntry.EventType eventType, CanalEntry.RowData rowData) {
        //do something...
        List<CanalEntry.Column> afterColumnsList = rowData.getAfterColumnsList();
        for (CanalEntry.Column column : afterColumnsList) {
            System.out.println(column.getName() + ":" + column.getValue());
        }
    }

    @UpdateListenPoint
    public void onEvent1(CanalEntry.RowData rowData) {
        //do something...
    }

    @DeleteListenPoint
    public void onEvent3(CanalEntry.EventType eventType) {
        //do something...
    }

    @ListenPoint(destination = "example", schema = "canal-test", table = {"t_user", "test_table"}, eventType = CanalEntry.EventType.UPDATE)
    public void onEvent4(CanalEntry.EventType eventType, CanalEntry.RowData rowData) {
        //do something...
    }
}