[大数据]docker安装canal

#数据库开启binlog
查看我之前的教程 docker 创建MySQL

docker cp mysql:/etc/mysql/mysql.conf.d/mysqld.cnf  /
vim /mysqld.cnf
i
##添加下面的内容
log-bin=mysql-bin
server-id=1

保存 重启mysql容器

安装canal

docker run -p  11111:11111   --name=canal -id canal/canal-server
##进入 canal 容器
docker exec  -it canal bash
#编辑
cd /home/admin/canal-server/conf/example/
vi instance.properties

保存 重启cannel容器

java客户端

##导入两个依赖
    <dependency>
      <groupId>com.alibaba.otter</groupId>
      <artifactId>canal.client</artifactId>
      <version>1.1.5</version>
    </dependency>

    <!-- Message、CanalEntry.Entry等来自此安装包 -->
    <dependency>
      <groupId>com.alibaba.otter</groupId>
      <artifactId>canal.protocol</artifactId>
      <version>1.1.5</version>
    </dependency>
    

测试用例

package com.zixi;

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.InvalidProtocolBufferException;

import java.net.InetSocketAddress;
import java.util.List;

/**
 * @创建人 jiangxiong
 * @创建时间 2021-09-05 22:14
 * @描述
 */
public class Simple {

    //Canal服务地址
    private static final String SERVER_ADDRESS = "服务地址";

    //Canal Server 服务端口号
    private static final Integer PORT = 11111;

    //目的地，其实Canal Service内部有一个队列,和配置文件中一致即可，参考【修改instance.properties】图中
    private static final String DESTINATION = "example";

    //用户名和密码，但是目前不支持，只能为空
    private static final String USERNAME = "";

    //用户名和密码，但是目前不支持，只能为空
    private static final String PASSWORD = "";

    public static void main(String[] args) {
        CanalConnector canalConnector = CanalConnectors.newSingleConnector(new InetSocketAddress(SERVER_ADDRESS, PORT), DESTINATION, USERNAME, PASSWORD);
        canalConnector.connect();
        //订阅所有消息
        canalConnector.subscribe(".*\\..*");
        //恢复到之前同步的那个位置
        canalConnector.rollback();

        for (; ; ) {
            //获取指定数量的数据，但是不做确认标记，下一次取还会取到这些信息
            Message message = canalConnector.getWithoutAck(100);
            //获取消息id
            long batchId = message.getId();
            if (batchId != -1) {
                System.out.println("msgId -> " + batchId);
                printEnity(message.getEntries());
                //提交确认
                //canalConnector.ack(batchId);
                //处理失败，回滚数据
                //canalConnector.rollback(batchId);
            }
        }
    }

    private static void printEnity(List<CanalEntry.Entry> entries) {
        for (CanalEntry.Entry entry : entries) {
            if (entry.getEntryType() != CanalEntry.EntryType.ROWDATA) {
                continue;
            }
            try {
                CanalEntry.RowChange rowChange = CanalEntry.RowChange.parseFrom(entry.getStoreValue());
                for (CanalEntry.RowData rowData : rowChange.getRowDatasList()) {
                    System.out.println(rowChange.getEventType());
                    switch (rowChange.getEventType()) {
                        //如果希望监听多种事件，可以手动增加case
                        case INSERT:
                            String tableName = entry.getHeader().getTableName();
                            //测试users表进行映射处
                            List<CanalEntry.Column> afterColumnsList = rowData.getAfterColumnsList();
                            System.out.println(afterColumnsList);
                            break;
                        case UPDATE:
                            List<CanalEntry.Column> afterColumnsList2 = rowData.getAfterColumnsList();
                            System.out.println("新插入的数据是：" + afterColumnsList2);
                            break;
                        case DELETE:
                            List<CanalEntry.Column> beforeColumnsList = rowData.getBeforeColumnsList();
                            System.out.println("被删除的数据是：" + beforeColumnsList);
                            break;
                        default:
                    }
                }
            } catch (InvalidProtocolBufferException e) {
                e.printStackTrace();
            }
        }

    }
}