From d0ac50d62637c457501f2fa31fa2bd082a8fcf77 Mon Sep 17 00:00:00 2001 From: agapple Date: Sun, 25 Nov 2018 20:48:24 +0800 Subject: [PATCH] Updated Canal Kafka RocketMQ QuickStart (markdown) --- Canal-Kafka-RocketMQ-QuickStart.md | 78 ++++++++++++++---------------- 1 file changed, 37 insertions(+), 41 deletions(-) diff --git a/Canal-Kafka-RocketMQ-QuickStart.md b/Canal-Kafka-RocketMQ-QuickStart.md index 346ed70..768cebf 100644 --- a/Canal-Kafka-RocketMQ-QuickStart.md +++ b/Canal-Kafka-RocketMQ-QuickStart.md @@ -46,6 +46,13 @@ canal.instance.master.address=192.168.1.20:3306 canal.instance.dbUsername = canal canal.instance.dbPassword = canal ... +# mq config +canal.mq.topic=example +canal.mq.partition=0 +# hash partition config +#canal.mq.partitionsNum=3 +#库名.表名: 唯一主键，多个表之间用逗号分隔 +#canal.mq.partitionHash=mytest.person:id,mytest.role:id ################################################# ``` @@ -57,55 +64,44 @@ canal.instance.dbPassword = canal # 可选项: tcp(默认), kafka, RocketMQ canal.serverMode = kafka # ... -``` -其他详细参数可参考[Canal AdminGuide](https://github.com/alibaba/canal/wiki/AdminGuide) - -#### c. 修改mq配置文件`vi /usr/local/canal/conf/mq.yml` - -``` # kafka/rocketmq 集群配置: 192.168.1.117:9092,192.168.1.118:9092,192.168.1.119:9092 -servers: 192.168.1.117:9092 -retries: 0 -batchSize: 16384 -lingerMs: 1 -bufferMemory: 33554432 - +canal.mq.servers = 127.0.0.1:6667 +canal.mq.retries = 0 +canal.mq.batchSize = 16384 +canal.mq.maxRequestSize = 1048576 +canal.mq.lingerMs = 1 +canal.mq.bufferMemory = 33554432 # Canal的batch size, 默认50K, 由于kafka最大消息体限制请勿超过1M(900K以下) -canalBatchSize: 50 +canal.mq.canalBatchSize = 50 # Canal get数据的超时时间, 单位: 毫秒, 空为不限超时 -canalGetTimeout: 100 -flatMessage: true - -canalDestinations: - - canalDestination: example - topic: example - partition: 1 -# #对应topic分区数量 -# partitionsNum: 3 -# partitionHash: -# #库名.表名: 唯一主键 -# mytest.person: id +canal.mq.canalGetTimeout = 100 +# 是否为flat json格式对象 +canal.mq.flatMessage = true +canal.mq.compressionType = none +canal.mq.acks = all ``` -## mq.yml参数说明 +## mq相关参数说明 | 参数名 | 参数说明 | 默认值 | |-------|-----------|-------| -| servers | kafka为bootstrap.servers
rocketMQ中为nameserver列表 | 127.0.0.1:6667 | -| retries | 发送失败重试次数 | 0 | -| batchSize | kafka为`ProducerConfig.BATCH_SIZE_CONFIG`
rocketMQ无意义 | 16384 | -| lingerMs | kafka为`ProducerConfig.LINGER_MS_CONFIG`
rocketMQ无意义 | 1 | -| bufferMemory | kafka为`ProducerConfig.BUFFER_MEMORY_CONFIG`
rocketMQ无意义 | 33554432 | -| producerGroup | kafka无意义
rocketMQ为ProducerGroup名 | Canal-Producer | -| canalBatchSize | 获取canal数据的批次大小 | 50 | -| canalGetTimeout | 获取canal数据的超时时间 | 100 | -| flatMessage | 是否为json格式
如果设置为false,对应MQ收到的消息为protobuf格式
需要通过CanalMessageDeserializer进行解码 | true | -| CanalDestinations | 通道定义，父级目录 | -- | -| CanalDestination | 通道名称 | 无 | -| topic | mq里的topic名 | 无 | -| partition | 单队列模式的分区下标， | 1 | -| partitionsNum | 散列模式的分区数 | 无 | -| partitionHash | 散列规则定义
库名.表名 : 唯一主键，比如mytest.person: id | 无 | +| 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`
rocketMQ无意义 | 1 | +| canal.mq.bufferMemory | kafka为`ProducerConfig.BUFFER_MEMORY_CONFIG`
rocketMQ无意义 | 33554432 | +| canal.mq.producerGroup | kafka无意义
rocketMQ为ProducerGroup名 | Canal-Producer | +| canal.mq.canalBatchSize | 获取canal数据的批次大小 | 50 | +| canal.mq.canalGetTimeout | 获取canal数据的超时时间 | 100 | +| canal.mq.flatMessage | 是否为json格式
如果设置为false,对应MQ收到的消息为protobuf格式
需要通过CanalMessageDeserializer进行解码 | true | +| --- | --- | --- | +| canal.mq.topic | mq里的topic名 | 无 | +| canal.mq.partition | 单队列模式的分区下标， | 1 | +| canal.mq.partitionsNum | 散列模式的分区数 | 无 | +| canal.mq.partitionHash | 散列规则定义
库名.表名 : 唯一主键，比如mytest.person: id | 无 | + +其他详细参数可参考[Canal AdminGuide](https://github.com/alibaba/canal/wiki/AdminGuide) ### 3.4 启动 ```