0
点赞
收藏
分享

微信扫一扫

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


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的值

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

2、修改kafka配置

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

2、修改 instance.propertios 配置文件

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

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、获取消息

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

四、MQ配置相关的参数

参数名

参数说明

默认值

canal.mq.servers

kafka为bootstrap.servers rocketMQ中为nameserver列表

127.0.0.1:6667

canal.mq.retries

发送失败重试次数

0

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代码

​​https://gitee.com/huan1993/spring-cloud-parent/tree/master/canal/canal-kafka-consumer​​

六、参考文章

​​1、canal 发送binlog到mq中性能测试​​​.
​​​2、canal发送消息到kafka中​​


举报

相关推荐

0 条评论