docker 部署canal 并且在java客户端消费

在Java客户端中使用Canal消费MySQL的变更数据，需要按照以下步骤进行操作：

1. 引入Canal的客户端依赖：

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

2. 使用Canal客户端连接Canal Server：

   CanalConnector connector = CanalConnectors.newSingleConnector(
       new InetSocketAddress("127.0.0.1", 11111), "example", "", "");
   connector.connect();
   connector.subscribe(".*\\..*");
   connector.rollback();

3. 获取变更数据：

   while (true) {
       Message message = connector.getWithoutAck(100);
       long batchId = message.getId();
       int size = message.getEntries().size();
       if (batchId == -1 || size == 0) {
           try {
               Thread.sleep(1000);
           } catch (InterruptedException e) {
               e.printStackTrace();
           }
       } else {
           // 解析变更数据
           printEntries(message.getEntries());
           connector.ack(batchId);
       }
   }

4. 解析变更数据：

   private static void printEntries(List<Entry> entries) {
       for (Entry entry : entries) {
           if (entry.getEntryType() == EntryType.TRANSACTIONBEGIN || entry.getEntryType() == EntryType.TRANSACTIONEND) {
               continue;
           }
           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);
           }
           EventType eventType = rowChange.getEventType();
           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));
           for (RowData rowData : rowChange.getRowDatasList()) {
               if (eventType == EventType.DELETE) {
                   printColumn(rowData.getBeforeColumnsList());
               } else if (eventType == EventType.INSERT) {
                   printColumn(rowData.getAfterColumnsList());
               } else {
                   System.out.println("-------> before");
                   printColumn(rowData.getBeforeColumnsList());
                   System.out.println("-------> after");
                   printColumn(rowData.getAfterColumnsList());
               }
           }
       }
   }

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

在Java客户端中使用Canal消费MySQL的变更数据，需要先在Canal Server中配置好MySQL的连接信息和表信息，然后在Java客户端中连接Canal Server并解析变更数据。具体操作还需要根据自己的实际情况进行调整。