canal同步mysql binlog数据到kafka

悠悠 2022-09-03 07:28 18阅读 0赞

### 文章目录 ###

*  canal简介
 *  安装canal
 *  写入kafka
 *   *  单条写入限制
     *  连接需要sasl鉴权的kafka

# canal简介 #

[canal官网简介][canal]

> canal，译意为水道/管道/沟渠，主要用途是基于 MySQL 数据库增量日志解析，提供增量数据订阅和消费。

这句介绍有几个关键字：**增量日志**，**增量数据订阅和消费**。

这里我们可以简单地把canal理解为一个用来**同步增量数据的一个工具**。

接下来我们看一张官网提供的示意图：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70]  
canal的工作原理就是**把自己伪装成MySQL slave，模拟MySQL slave的交互协议向MySQL Mater发送 dump协议，MySQL mater收到canal发送过来的dump请求，开始推送binary log给canal，然后canal解析binary log，再发送到存储目的地**，比如MySQL，Kafka，Elastic Search等等。

# 安装canal #

参考[canal github官网][canal github]进行安装：

**准备**

对于自建 MySQL , 需要先开启 Binlog 写入功能，配置 binlog-format 为 ROW 模式，my.cnf 中配置如下

[mysqld]
    log-bin=mysql-bin # 开启 binlog
    binlog-format=ROW # 选择 ROW 模式
    server_id=1 # 配置 MySQL replaction 需要定义，不要和 canal 的 slaveId 重复

授权 canal 链接 MySQL 账号具有作为 MySQL slave 的权限, 如果已有账户可直接 grant

CREATE USER canal IDENTIFIED BY 'canal';  
    GRANT SELECT, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO 'canal'@'%';
    -- GRANT ALL PRIVILEGES ON *.* TO 'canal'@'%' ;
    FLUSH PRIVILEGES;

**启动**

下载 canal, 访问 [release][] 页面 , 选择需要的包下载, 如以 1.1.4 版本为例

wget https://github.com/alibaba/canal/releases/download/canal-1.1.4/canal.deployer-1.1.4.tar.gz

**解压缩**

mkdir /tmp/canal
    tar zxvf canal.deployer-$version.tar.gz  -C /tmp/canal

解压完成后，进入 /tmp/canal 目录，可以看到如下结构

drwxr-xr-x 2 root root     4096 Aug  6 11:05 bin
    drwxr-xr-x 5 root root     4096 Aug  6 10:46 conf
    drwxr-xr-x 2 root root     4096 Jul 24 19:56 lib
    drwxrwxrwx 4 root root     4096 Jul 24 20:30 logs

配置修改

vi conf/example/instance.properties

## mysql serverId
    canal.instance.mysql.slaveId = 1234
    #position info，需要改成自己的数据库信息
    canal.instance.master.address = 127.0.0.1:3306 
    canal.instance.master.journal.name = 
    canal.instance.master.position = 
    canal.instance.master.timestamp = 
    #canal.instance.standby.address = 
    #canal.instance.standby.journal.name =
    #canal.instance.standby.position = 
    #canal.instance.standby.timestamp = 
    #username/password，需要改成自己的数据库信息
    canal.instance.dbUsername = canal  
    canal.instance.dbPassword = canal
    canal.instance.defaultDatabaseName =
    canal.instance.connectionCharset = UTF-8
    #table regex
    canal.instance.filter.regex = .*\\..*

canal.instance.connectionCharset 代表数据库的编码方式对应到 java 中的编码类型，比如 UTF-8，GBK , ISO-8859-1  
如果系统是1个 cpu，需要将 canal.instance.parser.parallel 设置为 false

**启动**

sh bin/startup.sh

查看 server 日志

[root@hostxxxxxxxxx canal]# tail -fn 99 logs/canal/canal.log
    2021-08-06 11:05:15.849 [main] INFO  com.alibaba.otter.canal.deployer.CanalStarter - ## start the canal server.
    2021-08-06 11:05:15.886 [main] INFO  com.alibaba.otter.canal.deployer.CanalController - ## start the canal server[x.x.x.x(x.x.x.x):11111]
    2021-08-06 11:05:17.237 [main] INFO  com.alibaba.otter.canal.deployer.CanalStarter - ## the canal server is running now ......

查看 instance 的日志

2021-08-07 09:54:52.226 [main] INFO  c.a.o.c.i.spring.support.PropertyPlaceholderConfigurer - Loading properties file from class path resource [canal.properties]
    2021-08-07 09:54:52.227 [main] INFO  c.a.o.c.i.spring.support.PropertyPlaceholderConfigurer - Loading properties file from class path resource [example/instance.properties]
    2021-08-07 09:54:52.746 [main] INFO  c.a.otter.canal.instance.spring.CanalInstanceWithSpring - start CannalInstance for 1-example
    2021-08-07 09:54:52.754 [main] WARN  c.a.o.canal.parse.inbound.mysql.dbsync.LogEventConvert - --> init table filter : ^.*\..*$
    2021-08-07 09:54:52.754 [main] WARN  c.a.o.canal.parse.inbound.mysql.dbsync.LogEventConvert - --> init table black filter :
    2021-08-07 09:54:52.878 [main] INFO  c.a.otter.canal.instance.core.AbstractCanalInstance - start successful....

canal默认的模式是tcp，此外还有kafka、RocketMQ，可以在canal.properties中看到：

# tcp, kafka, RocketMQ
    canal.serverMode = tcp

此时可以通过如下代码读取到binlog消息：

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 org.springframework.beans.factory.InitializingBean;
    import org.springframework.stereotype.Component;
    
    import java.net.InetSocketAddress;
    import java.text.SimpleDateFormat;
    import java.util.HashMap;
    import java.util.List;
    import java.util.Map;
    
    /** * @program: * @description: 连接cancel读取mysql binlog数据 * @author: zhangbin * @create: 2021-08-04 15:37 **/
    public class CannalClient implements InitializingBean { 
    
        private final static int BATCH_SIZE = 1000;
        private final static SimpleDateFormat sdr = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
    
        @Override
        public void afterPropertiesSet() throws Exception { 
            // 创建链接
            CanalConnector connector = CanalConnectors.newSingleConnector(new InetSocketAddress("yourip", 11111), "example", "", "");
            try { 
                //打开连接
                connector.connect();
                //订阅数据库表,全部表
                connector.subscribe(".*\\..*");
                //回滚到未进行ack的地方，下次fetch的时候，可以从最后一个没有ack的地方开始拿
                connector.rollback();
                while (true) { 
                    // 获取指定数量的数据
                    Message message = connector.getWithoutAck(BATCH_SIZE);
                    //获取批量ID
                    long batchId = message.getId();
                    //获取批量的数量
                    int size = message.getEntries().size();
                    //如果没有数据
                    if (batchId == -1 || size == 0) { 
                        try { 
                            //线程休眠2秒
                            Thread.sleep(2000);
                        } catch (InterruptedException e) { 
                            e.printStackTrace();
                        }
                    } else { 
                        //如果有数据,处理数据
                        printEntry(message.getEntries());
                    }
                    //进行 batch id 的确认。确认之后，小于等于此 batchId 的 Message 都会被确认。
                    connector.ack(batchId);
                }
            } catch (Exception e) { 
                e.printStackTrace();
            } finally { 
                connector.disconnect();
            }
        }
    
        /** * 打印canal server解析binlog获得的实体类信息 */
        private static void printEntry(List<CanalEntry.Entry> entrys) { 
            for (CanalEntry.Entry entry : entrys) { 
                Map<String, String> map = new HashMap<String, String>();
                if (entry.getEntryType() == CanalEntry.EntryType.TRANSACTIONBEGIN || entry.getEntryType() == CanalEntry.EntryType.TRANSACTIONEND) { 
                    //开启/关闭事务的实体类型，跳过
                    continue;
                }
                //RowChange对象，包含了一行数据变化的所有特征
                //比如isDdl 是否是ddl变更操作 sql 具体的ddl sql beforeColumns afterColumns 变更前后的数据字段等等
                CanalEntry.RowChange rowChage;
                try { 
                    rowChage = CanalEntry.RowChange.parseFrom(entry.getStoreValue());
                } catch (Exception e) { 
                    throw new RuntimeException("ERROR ## parser of eromanga-event has an error , data:" + entry.toString(), e);
                }
                //获取操作类型：insert/update/delete类型
                CanalEntry.EventType eventType = rowChage.getEventType();
                map.put("eventType", eventType.toString());
                map.put("schemaName", entry.getHeader().getSchemaName());
                map.put("tableName", entry.getHeader().getTableName());
                map.put("isDdl", rowChage.getIsDdl() == true ? "1" : "0");
                map.put("logfileName", entry.getHeader().getLogfileName());
                map.put("logfileOffset", String.valueOf(entry.getHeader().getLogfileOffset()));
                map.put("executeTime", sdr.format(entry.getHeader().getExecuteTime()));
    
                //判断是否是DDL语句
                if (rowChage.getIsDdl()) { 
                    System.out.println("================》;isDdl: true,sql:" + rowChage.getSql());
                    return;
                }
    
                //获取RowChange对象里的每一行数据，打印出来
                Map<String, String> dataMap = new HashMap<String, String>();
                Map<String, String> oldDataMap = new HashMap<String, String>();
    
                for (CanalEntry.RowData rowData : rowChage.getRowDatasList()) { 
                    //如果是删除语句
                    if (eventType == CanalEntry.EventType.DELETE) { 
                        printColumn(oldDataMap, rowData.getBeforeColumnsList());
                        //如果是新增语句
                    } else if (eventType == CanalEntry.EventType.INSERT) { 
                        printColumn(dataMap, rowData.getAfterColumnsList());
                        //如果是更新的语句
                    } else { 
                        //变更前的数据
                        printColumn(oldDataMap, rowData.getBeforeColumnsList());
                        //变更后的数据
                        printColumn(dataMap, rowData.getAfterColumnsList());
                    }
                }
                map.put("newData", dataMap == null ? "" : dataMap.toString());
                map.put("oldData", oldDataMap == null ? "" : oldDataMap.toString());
                System.out.println("===data===：" + map.toString());
            }
        }
    
        private static void printColumn(Map<String, String> dataMap, List<CanalEntry.Column> columns) { 
            for (CanalEntry.Column column : columns) { 
    // System.out.println(column.getName() + " : " + column.getValue() + " update=" + column.getUpdated());
                dataMap.put(column.getName(), column.getValue());
            }
        }
    
        public static void main(String[] args) throws Exception { 
            new CannalClient().afterPropertiesSet();
        }
    }

关闭

sh bin/stop.sh

# 写入kafka #

上面的启动配置为默认tcp模式启动，如果需要通过canal直接将binlog数据写入kafka可以更改canal.properties配置文件：

[root@hostxxxxxxx canal]# vim conf/canal.properties

# tcp, kafka, RocketMQ
    canal.serverMode = kafka
    
    ##################################################
    ######### MQ #############
    ##################################################
    canal.mq.servers = kafkaip1:9110,kafkaip2:9110,kafkaip3:9110
    # 需要sasl鉴权需要添加如下两个参数，同时更改start.sh添加启动配置项
    canal.mq.properties.security.protocol = SASL_PLAINTEXT
    canal.mq.properties.sasl.mechanism = PLAIN
    
    canal.mq.retries = 0
    canal.mq.batchSize = 16384
    #canal.mq.maxRequestSize = 1048576
    # 默认单条最大记录为1m，不满足需求，这里改为3m
    canal.mq.maxRequestSize = 3145728
    canal.mq.lingerMs = 100
    canal.mq.bufferMemory = 33554432
    canal.mq.canalBatchSize = 50
    canal.mq.canalGetTimeout = 100
    canal.mq.flatMessage = true
    # 默认写入topic不开启压缩，结果受限kafka单条写入大小，这里开启gzip压缩写入
    #canal.mq.compressionType = none
    canal.mq.compressionType = gzip
    canal.mq.acks = all
    #canal.mq.properties. =
    canal.mq.producerGroup = test
    # Set this value to "cloud", if you want open message trace feature in aliyun.
    canal.mq.accessChannel = local

更改instance.properties配置文件指定要写入的topic：

vim conf/example/instance.properties

# mq config
    # 配置要指定的topic名
    canal.mq.topic=mysql-binlog-data
    # dynamic topic route by schema or table regex
    #canal.mq.dynamicTopic=mytest1.user,mytest2\\..*,.*\\..*
    canal.mq.partition=0
    # hash partition config
    canal.mq.partitionsNum=3
    # 指定所有正则匹配到的表的主键(自动查找)为hash字段
    canal.mq.partitionHash=.*\\..*:$pk$

注意这里有两个要注意的点：

*  单条写入限制
 *  kafka写入需要sasl鉴权

## 单条写入限制 ##

上面的配置文件中更改了`canal.mq.maxRequestSize`的值，默认值为1m，可是binlog部分数据比较大会超过这个值，此处改为了3m。
    更改前报错如下：
    ```powershell
    java.lang.RuntimeException: java.util.concurrent.ExecutionException: org.apache.kafka.common.errors.RecordTooLargeException: The message is 2798129 bytes when serialized which is larger than the maximum request size you have configured with the max.request.size configuration.
    ```
    
    在更改了`canal.mq.maxRequestSize`的值后还是写入失败，原因是kafka对单条写入数据的大小有限制，次数更改了压缩方式为gzip（默认为none，不开启压缩）`canal.mq.compressionType = gzip`。

## 连接需要sasl鉴权的kafka ##

要写入的kafka开启了sasl验证，需要更改如下三个地方：

1.  canal.properties中新增两行配置，用来开启sasl验证模式：
    
        canal.mq.properties.security.protocol = SASL_PLAINTEXT
        canal.mq.properties.sasl.mechanism = PLAIN
2.  准备一个文件kafka\_client\_jaas.conf，放在config目录下
    
        vim conf/kafka_client_jaas.conf
    
        KafkaClient { 
             org.apache.kafka.common.security.plain.PlainLoginModule required
             username="yourusername"
             password="yourpassword";
        };
3.  修改启动脚本 start.sh ，配置 `-Djava.security.auth.login.config=/opt/canal/conf/kafka_client_jaas.conf`。 具体位置参考以下代码：
    
        if [ -n "$str" ]; then
         JAVA_OPTS="-server -Xms2048m -Xmx3072m -Xmn1024m -XX:SurvivorRatio=2 -XX:PermSize=96m -XX:MaxPermSize=256m -Xss256k -XX:-UseAdaptiveSizePolicy -XX:MaxTenuringThreshold=15 -XX:+DisableExplicitGC -XX:+UseConcMarkSweepGC -XX:+CMSParallelRemarkEnabled -XX:+UseCMSCompactAtFullCollection -XX:+UseFastAccessorMethods -XX:+UseCMSInitiatingOccupancyOnly -XX:+HeapDumpOnOutOfMemoryError -Djava.security.auth.login.config=/opt/canal/conf/kafka_client_jaas.conf "
        else
             JAVA_OPTS="-server -Xms1024m -Xmx1024m -XX:NewSize=256m -XX:MaxNewSize=256m -XX:MaxPermSize=128m "
        fi

如上配置，重新启动canal后，binlog会写入到对应kafka topic中了

canal quickstart：  
https://github.com/alibaba/canal/wiki/QuickStart

kafka 开启sasl验证，参考：  
https://www.cnblogs.com/maoxiangyi/p/12896377.html  
https://blog.csdn.net/u010416101/article/details/79562214?utm\_source=blogxgwz0

canal写入kafka/rocketmq配置说明：  
https://github.com/alibaba/canal/wiki/Canal-Kafka-RocketMQ-QuickStart

[canal]: https://github.com/alibaba/canal/wiki/%E7%AE%80%E4%BB%8B
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70]: /images/20220829/8cfc218f2a1c4a94b95fd9489ec1cab5.png
[canal github]: https://github.com/alibaba/canal/wiki/QuickStart
[release]: https://github.com/alibaba/canal/releases