数据同步方案canal客户端连接服务器之kafka异步模式

准备好kafka环境

先安装zookeeper 将下载好的zookeeper中的zoo_sample.cfg 修改为 zoo.cfg 修改 zoo.cfg 中的 dataDir=E:\zkkafka\zookeeper-3.4.14\data

新增环境变量: ZOOKEEPER_HOME: E:\zkkafka\zookeeper-3.4.14 (zookeeper目录) Path: 在现有的值后面添加 “;%ZOOKEEPER_HOME%\bin;”

运行zk zkServer.cmd

安装kafka 解压 kafka_2.13-2.4.0 改名为 kafka

修改 server.properties中的配置

log.dirs=E:\zkkafka\kafka\logs cmd 进入到该目录： cd E:\zkkafka\kafka .\bin\windows\kafka-server-start.bat .\config\server.properties

Canal配置更改

修改 example/instance.properties canal.mq.topic=canal-topic修改 canal.properties canal.serverMode = kafka canal.mq.servers = 127.0.0.1:9092

SpringBoot项目整合kafka

添加maven依赖 <!-- springBoot集成canal客户端 --> <dependency> <groupId>com.alibaba.otter</groupId> <artifactId>canal.client</artifactId> <version>1.1.2</version> </dependency> <!-- springBoot集成kafka --> <dependency> <groupId>org.springframework.kafka</groupId> <artifactId>spring-kafka</artifactId> </dependency>

2.配置application.yml文件

# kafka spring: kafka: # kafka服务器地址(可以多个) bootstrap-servers: 127.0.0.1:9092 consumer: # 指定一个默认的组名 group-id: kafka2 # earliest:当各分区下有已提交的offset时，从提交的offset开始消费；无提交的offset时，从头开始消费 # latest:当各分区下有已提交的offset时，从提交的offset开始消费；无提交的offset时，消费新产生的该分区下的数据 # none:topic各分区都存在已提交的offset时，从offset后开始消费；只要有一个分区不存在已提交的offset，则抛出异常 auto-offset-reset: earliest # key/value的反序列化 key-deserializer: org.apache.kafka.common.serialization.StringDeserializer value-deserializer: org.apache.kafka.common.serialization.StringDeserializer producer: # key/value的序列化 key-serializer: org.apache.kafka.common.serialization.StringSerializer value-serializer: org.apache.kafka.common.serialization.StringSerializer # 批量抓取 batch-size: 65536 # 缓存容量 buffer-memory: 524288 编写kafka消费者 @Component public class KafkaCanalListener { @KafkaListener(topics = "test-topic") public void receive(ConsumerRecord<?, ?> consumer) { System.out.println("topic名称:" + consumer.topic() + ",key:" + consumer.key() + "," + "分区位置:" + consumer.partition() + ", 下标" + consumer.offset() + "," + consumer.value()); String json = (String) consumer.value(); JSONObject jsonObject = JSONObject.parseObject(json); String sqlType = jsonObject.getString("type"); JSONArray data = jsonObject.getJSONArray("data"); System.out.println("表名为:" + jsonObject.getString("table") + ",sql类型为：" + sqlType); if ("UPDATE".equals(sqlType) || "INSERT".equals(sqlType)) { System.out.println(data); return; } if ("DELETE".equals(sqlType)) { System.out.println(data); } } }