docker安装Canal，开启MySQL binlog ，连接Java，监控MySQL变化

以下是一个简化的Docker安装Canal并配置MySQL binlog，连接Java应用，并监控MySQL变化的例子。

首先，你需要有一个docker-compose.yml文件来定义Canal服务和MySQL服务。

version: '3'
services:
  mysql:
    image: mysql:5.7
    environment:
      MYSQL_ROOT_PASSWORD: 123456
      MYSQL_DATABASE: testdb
    command: --server-id=1 --log-bin=mysql-bin --binlog-format=ROW
 
  canal:
    image: canal/canal-server:v1.1.6
    links:
      - mysql
    environment:
      canal.destinations: test
      canal.instance.master.address: mysql:3306
      canal.instance.dbUsername: root
      canal.instance.dbPassword: 123456
    command: --auto-scan=false --deployer=canal.deployers.example.CanalLauncher

在这个例子中，我们定义了两个服务：mysql和canal。MySQL服务配置了环境变量和命令行参数来启用binlog。Canal服务配置了与MySQL数据库的连接信息。

接下来，你可以使用Docker Compose来启动服务：

docker-compose up -d

Canal现在会监控MySQL的变化，并且可以通过Java应用来接收这些变化。你可以使用Canal提供的客户端库（例如：canal-client-1.1.6-SNAPSHOT.jar）来连接Canal服务，并处理接收到的数据。

以下是一个简单的Java代码示例，用于连接Canal服务并打印收到的数据变化：

import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.client.CanalConnectors;
import com.alibaba.otter.canal.protocol.Message;
import com.alibaba.otter.canal.protocol.CanalEntry;
 
public class SimpleCanalClientExample {
    public static void main(String args[]) {
        // 连接Canal服务
        CanalConnector connector = CanalConnectors.newSingleConnector(
            new InetSocketAddress(AddressUtils.getHostIp(),
            11111), "test", "", "");
 
        int batchSize = 1000;
        try {
            connector.connect();
            connector.subscribe(".*\\..*");
            connector.rollback();
            while (true) {
                Message message = connector.getWithoutAck(batchSize); // 获取指定数量的数据
                long batchId = message.getId();
                if (batchId == -1 || message.getEntries().isEmpty()) {
                    // 没有数据，休眠一会儿
                    Thread.sleep(1000);
                } else {
                    dataHandle(message.getEntries());
                    connector.ack(batchId); // 确认消息消费成功
                }
            }
        } catch (Exception e) {
            e.printStackTrace();