接过火炬，升级canal-client-springboot-starter，一个支持rabbitmq的CanalClient_canal-spring-boot-starter

二、最终方案

---

参考canal-client-springboot-starter自己构建了一个easy-canal-client的项目。

源码已开源：https://gitee.com/cowboy2014/easy-canal-client.git

架构示意图如下：

2-1 canal server升级1.1.5

canal需要升级为1.1.5，canal把binlog数据解析完成后，就把数据直接投递给rabbitmq了——1.1.5可以把数据直接投递给rabbitmq。

canal.properties参考配置：

tcp, kafka, rocketMQ, rabbitMQ

canal.serverMode = rabbitMQ

##################################################

######### RabbitMQ #############

##################################################

rabbitmq.host = 172.16.150.11

rabbitmq.virtual.host = /

rabbitmq.exchange = canal-exchange

rabbitmq.username = canal

rabbitmq.password = canal%123

rabbitmq.deliveryMode =

conf/example/instance.properties参考配置：

position info

canal.instance.master.address=172.16.150.12:3306

username/password

canal.instance.dbUsername=canal

canal.instance.dbPassword=canal%123

table regex

canal.instance.filter.regex=lpm-center\.lpm_(park|company|store|route)

mq config

canal.mq.topic=shangwt

2-2 业务模块集成Rabbitmq

业务模块直接集成rabbitmq，作为消费者进行数据的解析、同步。

2-2 使用方法

1. 新建Handler，用@CanalTable注解标注

@Component

@CanalTable(value = “lpm_park”)

@Slf4j

public class ParkHandler implements EntryHandler {

@Resource

private ElSearchParkServiceImpl elSearchParkService;

@Resource

private EsIndexes esIndexes;

@Override

public void insert(Park park) {

log.info(“insert message {}”, park);

try {

elSearchParkService.synchronous(esIndexes.getPark(), park.getId());

} catch (IOException e) {

log.error(“es insert wrong!”);

}

}

@Override

public void update(Park before, Park park) {

try {

if (ObjectUtil.isNotEmpty(before.getDeleted()) && !before.getDeleted().equals(park.getDeleted()) && park.getDeleted() == 1){

this.delete(park);

}

if (ObjectUtil.isNotEmpty(before.getDeleted()) && !before.getDeleted().equals(park.getDeleted()) && park.getDeleted() == 0){

this.insert(park);

}

elSearchParkService.synchronous(esIndexes.getPark(), park.getId());

} catch (IOException e) {

log.error(“es insert wrong!”);

}

log.info(“update after {}”, park);

}

@Override

public void delete(Park park) {

log.info(“delete {}”, park);

elSearchParkService.deleteById(esIndexes.getPark(), park.getId().intValue());

}

}

2. rabbitmq消费者端监听队列，调用对应的Handler进行消息消费

/**

• canal消息同步

• @param info

*/

@RabbitListener(queues = {“shangwtQueue”},containerFactory = “multiListenerContainer”)

public void consumeMsg(FlatMessage info){

try {
自我介绍一下，小编13年上海交大毕业，曾经在小公司待过，也去过华为、OPPO等大厂，18年进入阿里一直到现在。

深知大多数Java工程师，想要提升技能，往往是自己摸索成长或者是报班学习，但对于培训机构动则几千的学费，着实压力不小。自己不成体系的自学效果低效又漫长，而且极易碰到天花板技术停滞不前！

因此收集整理了一份《2024年Java开发全套学习资料》，初衷也很简单，就是希望能够帮助到想自学提升又不知道该从何学起的朋友，同时减轻大家的负担。

既有适合小白学习的零基础资料，也有适合3年以上经验的小伙伴深入学习提升的进阶课程，基本涵盖了95%以上Java开发知识点，真正体系化！

由于文件比较大，这里只是将部分目录截图出来，每个节点里面都包含大厂面经、学习笔记、源码讲义、实战项目、讲解视频，并且会持续更新！

如果你觉得这些内容对你有帮助，可以扫码获取！！（备注Java获取）

最后

《一线大厂Java面试题解析+核心总结学习笔记+最新讲解视频+实战项目源码》，点击传送门即可获取！
5254671a72faed303032d36.jpg" alt=“img” style=“zoom: 33%;” />

最后

[外链图片转存中…(img-ACw14jMD-1712445312075)]

[外链图片转存中…(img-TYBoc5J9-1712445312075)]
《一线大厂Java面试题解析+核心总结学习笔记+最新讲解视频+实战项目源码》，点击传送门即可获取！