docker部署canal

• 一、部署mysql
• • 1.直接部署mysql8
  • 2.验证一下是否开启binlog及级别
• 二、部署canal
• • 1.首先run一个canal/canal-server:latest拷贝出配置文件
  • 2.需要着重注意启动内存的修改
  • 3.instance.properties修改
  • 4.启动canal
• 三、client-demo
• • 1.demo
  • 2.pom
  • 3.效果

8默认开启binlog级别为Row
部署时需要注意将配置文件和binlog文件挂载出来,如果binlog文件不挂载出来会读不到

docker run --name mysql -p 10006:3306 
-e MYSQL_ROOT_PASSWORD=123456 
-v /home/config/canal/mysql/my.cnf:/etc/mysql/my.cnf 
-v /home/config/canal/mysql/mysql:/var/lib/mysql 
-d mysql:latest --lower_case_table_names=1

show variables like 'log_bin'

show variables like 'binlog_format'

docker run --name canal -d canal/canal-server:latest

拷贝文件

docker cp canal:/home/admin/canal-server/conf/canal.properties ./
docker cp canal:/home/admin/canal-server/conf/example/instance.properties ./
docker cp canal:/home/admin/canal-server/bin/startup.sh ./

canal默认启动需要内存超过1g,我的云主机不具备启动条件,在容器退出前查看log下的err日志发现内存不足,所以修改了启动文件startup.sh,在前面已经挂载出来部署时候可根据实际情况修改

docker run --name canal -p 11111:11111 -d 
-v /home/config/canal/instance.properties:/home/admin/canal-server/conf/example/instance.properties 
-v /home/config/canal/canal.properties:/home/admin/canal-server/conf/canal.properties 
-v /home/config/canal/startup.sh:/home/admin/canal-server/bin/startup.sh 
canal/canal-server:latest

package com.example.canalclentdemo.demo;

import com.alibaba.fastjson2.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 org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;
import java.net.InetSocketAddress;
import java.util.List;


@RestController
public class CanalMain {
    private final String destination = "example";
    private final String ip = "127.0.1.1";

    @GetMapping("/test")
    public void test() throws InvalidProtocolBufferException {
        CanalConnector connector = CanalConnectors.newSingleConnector(new InetSocketAddress(ip,
                11111), destination, "", "");

        //help socket
        while (true){
            connector.connect();
            connector.subscribe("canal_demo.*");
            Message message = connector.get(1000);
            List entries = message.getEntries();
            //有数据变化
            if (entries.size() > 0){
                for (CanalEntry.Entry s : entries) {
                    String tableName = s.getHeader().getTableName();
                    CanalEntry.EntryType entryType = s.getEntryType();
                    ByteString storeValue = s.getStoreValue();
                    if (CanalEntry.EntryType.ROWDATA.equals(entryType)){
                        CanalEntry.RowChange v = CanalEntry.RowChange.parseFrom(storeValue);
                            //当前事件操作类型
                        CanalEntry.EventType eventType = v.getEventType();
                            //获取数据集
                        List rowDataList = v.getRowDatasList();
                        rowDataList.forEach(rd ->{
                            JSONObject before = new JSONObject();
                            JSONObject After = new JSONObject();
                            for (CanalEntry.Column column : rd.getBeforeColumnsList()) {
                                before.put(column.getName(),column.getValue());
                            }
                            for (CanalEntry.Column column : rd.getAfterColumnsList()) {
                                After.put(column.getName(),column.getValue());
                            }
                            //TODO 操作数据
                            System.out.println("tableName="+tableName+" eventType="+eventType+" before="+before+" After="+After);
                        });
                    }
                }

            }
        }
    }
}

        
            com.alibaba.otter
            canal.client
            1.1.6
        

        
            com.alibaba.otter
            canal.protocol
            1.1.6