From ff2f783292f9a4f89c9e8e4c4ef49ff2e5521faf Mon Sep 17 00:00:00 2001 From: agapple Date: Tue, 17 May 2022 16:44:26 +0800 Subject: [PATCH] Updated Canal Kafka RocketMQ QuickStart (markdown) --- Canal-Kafka-RocketMQ-QuickStart.md | 52 ++++++++++++++---------------- 1 file changed, 25 insertions(+), 27 deletions(-) diff --git a/Canal-Kafka-RocketMQ-QuickStart.md b/Canal-Kafka-RocketMQ-QuickStart.md index bb793e0..3c64bdc 100644 --- a/Canal-Kafka-RocketMQ-QuickStart.md +++ b/Canal-Kafka-RocketMQ-QuickStart.md @@ -2,6 +2,8 @@ canal 1.1.1版本之后, 默认支持将canal server接收到的binlog数据直接投递到MQ, 目前默认支持的MQ系统有: * kafka: [https://github.com/apache/kafka](https://github.com/apache/kafka) * RocketMQ : [https://github.com/apache/rocketmq](https://github.com/apache/rocketmq) +* RabbitMQ : [https://github.com/rabbitmq/rabbitmq-server](https://github.com/rabbitmq/rabbitmq-server) +* pulsarmq : [https://github.com/apache/pulsar](https://github.com/apache/pulsar) ## 环境版本 * 操作系统：CentOS release 6.6 (Final) @@ -17,6 +19,8 @@ canal 1.1.1版本之后, 默认支持将canal server接收到的binlog数据直 * Kafka安装参考：[Kafka QuickStart](https://github.com/alibaba/canal/wiki/Kafka-QuickStart) * RocketMQ安装参考：[RocketMQ QuickStart](https://rocketmq.apache.org/docs/quick-start) +* RabbitMQ安装参考：[RabbitMQ QuickStart](https://www.rabbitmq.com/download.html) +* PulsarMQ安装参考：[PulsarMQ QuickStart](https://pulsar.apache.org/docs/2.2.1/getting-started-docker/) ## 三、 安装canal.server @@ -28,8 +32,8 @@ canal 1.1.1版本之后, 默认支持将canal server接收到的binlog数据直 ``` mkdir -p /usr/local/canal -cp canal.deployer-1.1.1.tar.gz /usr/local/canal -tar -zxvf canal.deployer-1.1.1.tar.gz +cp canal.deployer-1.1.6.tar.gz /usr/local/canal +tar -zxvf canal.deployer-1.1.6.tar.gz ``` ### 3.3 配置修改参数 @@ -63,28 +67,16 @@ canal.mq.partition=0 #### b. 修改canal 配置文件`vi /usr/local/canal/conf/canal.properties ` ``` # ... -# 可选项: tcp(默认), kafka, RocketMQ +# 可选项: tcp(默认), kafka,RocketMQ,rabbitmq,pulsarmq canal.serverMode = kafka # ... -# kafka/rocketmq 集群配置: 192.168.1.117:9092,192.168.1.118:9092,192.168.1.119:9092 -canal.mq.servers = 127.0.0.1:6667 -canal.mq.retries = 0 -# flagMessage模式下可以调大该值, 但不要超过MQ消息体大小上限 -canal.mq.batchSize = 16384 -canal.mq.maxRequestSize = 1048576 -# flatMessage模式下请将该值改大, 建议50-200 -canal.mq.lingerMs = 1 -canal.mq.bufferMemory = 33554432 + # Canal的batch size, 默认50K, 由于kafka最大消息体限制请勿超过1M(900K以下) canal.mq.canalBatchSize = 50 # Canal get数据的超时时间, 单位: 毫秒, 空为不限超时 canal.mq.canalGetTimeout = 100 # 是否为flat json格式对象 canal.mq.flatMessage = false -canal.mq.compressionType = none -canal.mq.acks = all -# kafka消息投递是否使用事务 -canal.mq.transaction = false ``` ## mq相关参数说明 (<=1.1.4版本) @@ -127,9 +119,9 @@ canal.mq.transaction = false | 参数名 | 参数说明 | 默认值 | |------|-----------|-------| -| canal.aliyun.accessKey | 阿里云ak | 空值 | -| canal.aliyun.secretKey | 阿里云sk | 空值 | -| canal.aliyun.uid | 阿里云uid | 空值 | +| canal.aliyun.accessKey | 阿里云ak | 无 | +| canal.aliyun.secretKey | 阿里云sk | 无 | +| canal.aliyun.uid | 阿里云uid | 无 | | canal.mq.flatMessage | 是否为json格式
如果设置为false,对应MQ收到的消息为protobuf格式
需要通过CanalMessageDeserializer进行解码 | false | | canal.mq.canalBatchSize | 获取canal数据的批次大小 | 50 | | canal.mq.canalGetTimeout | 获取canal数据的超时时间 | 100 | @@ -153,26 +145,32 @@ canal.mq.transaction = false |------|-----------|-------| | rocketmq.producer.group | rocketMQ为ProducerGroup名 | test | | rocketmq.enable.message.trace | 是否开启message trace | false | -| rocketmq.customized.trace.topic | message trace的topic | 空值 | -| rocketmq.namespace | rocketmq的namespace | 空值 | +| rocketmq.customized.trace.topic | message trace的topic | 无 | +| rocketmq.namespace | rocketmq的namespace | 无 | | rocketmq.namesrv.addr | rocketmq的namesrv地址 | 127.0.0.1:9876 | | rocketmq.retry.times.when.send.failed | 重试次数 | 0 | | rocketmq.vip.channel.enabled | rocketmq是否开启vip channel | false | | rocketmq.tag | rocketmq的tag配置 | 空值 | | --- | --- | --- | -| canal.mq.vhost= | rabbitMQ配置 | 无 | -| canal.mq.exchange= | rabbitMQ配置 | 无 | -| canal.mq.username= | rabbitMQ配置 | 无 | -| canal.mq.password= | rabbitMQ配置 | 无 | -| canal.mq.aliyunuid= | rabbitMQ配置 | 无 | +| rabbitmq.host | rabbitMQ配置 | 无 | +| rabbitmq.virtual.host | rabbitMQ配置 | 无 | +| rabbitmq.exchange | rabbitMQ配置 | 无 | +| rabbitmq.username | rabbitMQ配置 | 无 | +| rabbitmq.password | rabbitMQ配置 | 无 | +| rabbitmq.deliveryMode | rabbitMQ配置 | 无 | +| --- | --- | --- | +| pulsarmq.serverUrl | pulsarmq配置 | 无 | +| pulsarmq.roleToken | pulsarmq配置 | 无 | +| pulsarmq.topicTenantPrefix | pulsarmq配置 | 无 | | --- | --- | --- | | canal.mq.topic | mq里的topic名 | 无 | | canal.mq.dynamicTopic | mq里的动态topic规则, 1.1.3版本支持 | 无 | | canal.mq.partition | 单队列模式的分区下标， | 1 | +| canal.mq.enableDynamicQueuePartition | 动态获取MQ服务端的分区数,如果设置为true之后会自动根据topic获取分区数替换canal.mq.partitionsNum的定义,目前主要适用于RocketMQ | false | | canal.mq.partitionsNum | 散列模式的分区数 | 无 | +| canal.mq.dynamicTopicPartitionNum | mq里的动态队列分区数,比如针对不同topic配置不同partitionsNum | 无 | | canal.mq.partitionHash | 散列规则定义
库名.表名 : 唯一主键，比如mytest.person: id
1.1.3版本支持新语法，见下文 | 无 | - ### canal.mq.dynamicTopic 表达式说明 canal 1.1.3版本之后, 支持配置格式：schema 或 schema.table，多个配置之间使用逗号或分号分隔