Canal Server发送binlog消息到Kafka消息队列中

• ​​一、背景​​
• ​​二、需要修改的地方​​

• ​​1、canal.properties 配置文件修改​​

• ​​1、修改canal.serverMode的值​​
• ​​2、修改kafka配置​​

• ​​2、修改 instance.propertios 配置文件​​
• ​​3、canal发消息到mq性能优化​​

• ​​三、kafka接收消息​​

• ​​1、canal 发送过来的消息​​
• ​​2、监听消息​​
• ​​3、获取消息​​

• ​​四、MQ配置相关的参数​​
• ​​五、MQ接收binlog代码​​
• ​​六、参考文章​​

一、背景

在​​上一篇文章​​中，我们使用 Canal Admin 搭建了Canal Server 集群，在这篇文章中，我们使用上篇文章的基础，将消息发送到kafka消息队列中。

二、需要修改的地方

以下 配置文件的修改，都是在 Canal Admin 上修改的。

1、canal.properties 配置文件修改

1、修改canal.serverMode的值

2、修改kafka配置

2、修改 instance.propertios 配置文件

3、canal发消息到mq性能优化

影响性能的几个参数：

1. ​

   ​canal.instance.memory.rawEntry = true​

   ​ (表示是否需要提前做序列化，非flatMessage场景需要设置为true)
2. ​

   ​canal.mq.flatMessage = false​

   ​(false代表二进制协议，true代表使用json格式，二进制协议有更好的性能)
3. ​

   ​canal.mq.dynamicTopic​

   ​ (动态topic配置定义，可以针对不同表设置不同的topic，在flatMessage模式下可以提升并行效率)
4. ​

   ​canal.mq.partitionsNum/canal.mq.partitionHash​

   ​ (分区配置，对写入性能有反作用，不过可以提升消费端的吞吐)

参考链接：​​https://github.com/alibaba/canal/wiki/Canal-MQ-Performance​​

三、kafka接收消息

1、canal 发送过来的消息

/**
 * canal 发送过来的消息
 *
 * @author huan.fu 2021/9/2 - 下午4:06
 */
@Getter
@Setter
@ToString
public class CanalMessage {

    /**
     * 测试得出 同一个事物下产生多个修改，这个id的值是一样的。
     */
    private Integer id;

    /**
     * 数据库或schema
     */
    private String database;
    /**
     * 表名
     */
    private String table;
    /**
     * 主键字段名
     */
    private List<String> pkNames;
    /**
     * 是否是ddl语句
     */
    private Boolean isDdl;
    /**
     * 类型:INSERT/UPDATE/DELETE
     */
    private String type;
    /**
     * binlog executeTime, 执行耗时
     */
    private Long es;
    /**
     * dml build timeStamp, 同步时间
     */
    private Long ts;
    /**
     * 执行的sql,dml sql为空
     */
    private String sql;
    /**
     * 数据列表
     */
    private List<Map<String, Object>> data;
    /**
     * 旧数据列表,用于update,size和data的size一一对应
     */
    private List<Map<String, Object>> old;
}      

2、监听消息

@Component
@Slf4j
public class KafkaConsumer {

    @KafkaListener(topics = "customer", groupId = "canal-kafka-springboot-001", concurrency = "5")
    public void consumer(ConsumerRecord<String, String> record, Acknowledgment ack) throws InterruptedException {
        log.info(Thread.currentThread().getName() + ":" + LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss")) + "接收到kafka消息,partition:" + record.partition() + ",offset:" + record.offset() + "value:" + record.value());

        CanalMessage canalMessage = JSON.parseObject(record.value(), CanalMessage.class);

        log.info("\r=================================");
        log.info("接收到的原始 canal message为: {}", record.value());
        log.info("转换成Java对象后转换成Json为 : {}", JSON.toJSONString(canalMessage));

        ack.acknowledge();
    }
}      

3、获取消息

四、MQ配置相关的参数

参数名	参数说明	默认值
canal.mq.servers	kafka为bootstrap.servers rocketMQ中为nameserver列表	127.0.0.1:6667
canal.mq.retries	发送失败重试次数
canal.mq.batchSize	kafka为​ ​ProducerConfig.BATCH_SIZE_CONFIG​ ​ rocketMQ无意义	16384
canal.mq.maxRequestSize	kafka为​ ​ProducerConfig.MAX_REQUEST_SIZE_CONFIG​ ​ rocketMQ无意义	1048576
canal.mq.lingerMs	kafka为​ ​ProducerConfig.LINGER_MS_CONFIG​ ​ , 如果是flatMessage格式建议将该值调大, 如: 200 rocketMQ无意义	1
canal.mq.bufferMemory	kafka为​ ​ProducerConfig.BUFFER_MEMORY_CONFIG​ ​ rocketMQ无意义	33554432
canal.mq.acks	kafka为​ ​ProducerConfig.ACKS_CONFIG​ ​ rocketMQ无意义	all
canal.mq.kafka.kerberos.enable	kafka为​ ​ProducerConfig.ACKS_CONFIG​ ​ rocketMQ无意义	false
canal.mq.kafka.kerberos.krb5FilePath	kafka kerberos认证 rocketMQ无意义	…/conf/kerberos/krb5.conf
canal.mq.kafka.kerberos.jaasFilePath	kafka kerberos认证 rocketMQ无意义	…/conf/kerberos/jaas.conf
canal.mq.producerGroup	kafka无意义 rocketMQ为ProducerGroup名	Canal-Producer
canal.mq.accessChannel	kafka无意义 rocketMQ为channel模式，如果为aliyun则配置为cloud	local
—	—	—
canal.mq.vhost=	rabbitMQ配置	无
canal.mq.exchange=	rabbitMQ配置	无
canal.mq.username=	rabbitMQ配置	无
canal.mq.password=	rabbitMQ配置	无
canal.mq.aliyunuid=	rabbitMQ配置	无
—	—	—
canal.mq.canalBatchSize	获取canal数据的批次大小	50
canal.mq.canalGetTimeout	获取canal数据的超时时间	100
canal.mq.parallelThreadSize	mq数据转换并行处理的并发度	8
canal.mq.flatMessage	是否为json格式 如果设置为false,对应MQ收到的消息为protobuf格式 需要通过CanalMessageDeserializer进行解码	false
—	—	—
canal.mq.topic	mq里的topic名	无
canal.mq.dynamicTopic	mq里的动态topic规则, 1.1.3版本支持	无
canal.mq.partition	单队列模式的分区下标，	1
canal.mq.partitionsNum	散列模式的分区数	无
canal.mq.partitionHash	散列规则定义 库名.表名 : 唯一主键，比如mytest.person: id 1.1.3版本支持新语法，见下文

参考文档：​​https://github.com/alibaba/canal/wiki/Canal-Kafka-RocketMQ-QuickStart​​

五、MQ接收binlog代码

六、参考文章