docker 部署canal 配置kafka

mysql开启binlog这里使用的是 mysql 5.7.32 ，server-id可以自定义，和后面配置的canal里面的slaveId要不一样# binloglog-bin=mysql-binbinlog_format=ROWserver-id=33081修改完成后，需要重启mysql服务show variables like 'log_bin';返回 ON 表示 binlog 启动成功±-

zhaoyang10

3799人浏览 · 2022-04-11 11:08:10

zhaoyang10 · 2022-04-11 11:08:10 发布

mysql开启binlog
这里使用的是 mysql 5.7.32 ，server-id可以自定义，和后面配置的canal里面的slaveId要不一样

# binlog
log-bin=mysql-bin
binlog_format=ROW
server-id=33081

修改完成后，需要重启mysql服务

show variables like 'log_bin';

返回 ON 表示 binlog 启动成功

±--------------±------+
| Variable_name | Value |
±--------------±------+
| log_bin | ON |
±--------------±------+
mysql创建canal用户

grant SELECT, REPLICATION SLAVE, REPLICATION CLIENT on *.* to 'canal'@'%' identified by "canal";
flush privileges;
show grants for 'canal'@'%%';

dokcer 部署 canal

mkdir /data0/canal

1.首先docker拉去canal镜像,这里不加版本号，默认为latest

docker pull canal/canal-server

2.查看镜像是否拉去成功

docker images

3.拉去完成后，先启动下canal，主要是为了从里面copy出配置文件

#启动镜像

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

#进入容器查看配置文件路径

docker exec -it canal bash

#找到文件位置后 exit退出容器将容器内部文件copy到外部

docker cp canal:/home/admin/canal-server/conf/canal.properties /data0/canal
docker cp canal:/home/admin/canal-server/conf/example/instance.properties /data0/canal

4.文件copy完成后主要是修改instance这个文件。第一个红框是你需要监听数据库的地址和端口；第二个红框是你数据库的用户和密码，这个用户信息一定是要有全部权限的用户，非root用户；第三个是匹配数据表的规则，我这里默认为全部表
在这里插入图片描述

5.修改完成后，将之前的canal容器关闭，重新起一个新的容器.

#关闭容器

docker stop canal

#移除容器

docker rm canal

#启动新的这里-v是将外部的文件挂载到容器内部这样就不用每次启动都要配置参数了

docker run --name canal -p 11111:11111 -d \
-v /data0/canal/instance.properties:/home/admin/canal-server/conf/example/instance.properties \
-v /data0/canal/canal.properties:/home/admin/canal-server/conf/canal.properties \
-v /data0/canal/logs/:/home/admin/canal-server/logs/ \
canal/canal-server

注意: 需要注释掉,不然修改mysql数据时,没有向java客户端返回响应

canal.properties

在这里插入图片描述
insert操作：


{
    "data":[
        {
            "id":"338",
            "city":"成都",
            "province":"四川省"
        }
    ],
    "database":"qfbap_ods",
    "es":1583394964000,
    "id":2,
    "isDdl":false,
    "mysqlType":{
        "id":"int(11)",
        "city":"varchar(256)",
        "province":"varchar(256)"
    },
    "old":null,
    "pkNames":[
        "id"
    ],
    "sql":"",
    "sqlType":{
        "id":4,
        "city":12,
        "province":12
    },
    "table":"code_city",
    "ts":1583394964361,
    "type":"INSERT"
}

update操作：

{
    "data":[
        {
            "id":"338",
            "city":"绵阳市",
            "province":"四川省"
        }
    ],
    "database":"qfbap_ods",
    "es":1583395177000,
    "id":3,
    "isDdl":false,
    "mysqlType":{
        "id":"int(11)",
        "city":"varchar(256)",
        "province":"varchar(256)"
    },
    "old":[
        {
            "city":"成都"
        }
    ],
    "pkNames":[
        "id"
    ],
    "sql":"",
    "sqlType":{
        "id":4,
        "city":12,
        "province":12
    },
    "table":"code_city",
    "ts":1583395177408,
    "type":"UPDATE"
}

delete操作：

{
    "data":[
        {
            "id":"338",
            "city":"绵阳市",
            "province":"四川省"
        }
    ],
    "database":"qfbap_ods",
    "es":1583395333000,
    "id":4,
    "isDdl":false,
    "mysqlType":{
        "id":"int(11)",
        "city":"varchar(256)",
        "province":"varchar(256)"
    },
    "old":null,
    "pkNames":[
        "id"
    ],
    "sql":"",
    "sqlType":{
        "id":4,
        "city":12,
        "province":12
    },
    "table":"code_city",
    "ts":1583395333208,
    "type":"DELETE"
}

JSON日志格式解释
data：最新的数据，为JSON数组，如果是插入则表示最新插入的数据，如果是更新，则表示更新后的最新数据，如果是删除，则表示被删除的数据
database：数据库名称
es：事件时间，13位的时间戳
id：事件操作的序列号，1,2,3…
isDdl：是否是DDL操作
mysqlType：字段类型
old：旧数据
pkNames：主键名称
sql：SQL语句
sqlType：是经过canal转换处理的，比如unsigned int会被转化为Long，unsigned long会被转换为BigDecimal
table：表名
ts：日志时间
type：操作类型，比如DELETE，UPDATE，INSERT