当前位置：首页 > news >正文

云岚到家，使用Elasticsearch实现服务的搜索功能，使用Canal+MQ完成服务信息与ES索引同步。MQ

news 文章来源：https://blog.csdn.net/2201_75600005/article/details/142767469 2025/5/14 14:30:19

为什么使用elasticsearch?数据很多么？

项目使用Elasticsearch是实现了门户上对服务的搜索。

平台上的服务数据是并不是很多，全国所有区域下的服务信息加一起几千条，之所以使用Elasticsearch是因为：

1、公司架构师在系统架构时考虑几年后的数据及对全文检索使用的需求使用了Elasticsearch.

2、对服务信息进行搜索使用的是全文检索方式，虽然MySQL也支持全文检索但是我们这个接口是面向 C端用户且对接口性能有要求，所以使用了ES。

3、虽然现在数据量不大考虑几年后的数据量增长问题，我们使用了Elasticsearch。

4、在项目中除了通过关键字搜索服务信息，还有根据地理坐标进行搜索，使用Elasticsearch也考虑了这一点。

实现目标：

对于服务类型和服务项的名称进行索引，当搜索服务项名称时，展示出该服务项，当搜索服务类型时，展示出该类型下的所有服务项。

方案一

因为操作es和数据库是分布式事务，无法控制一致性。

方案二

使用Canal+MQ

Canal可与很多数据源进行对接，将数据由MySQL同步到ES、MQ、DB等各个数据源。

Canal的意思是水道/管道/沟渠，它相当于一个数据管道，通过解析MySQL的binlog日志完成数据同步工作。

要理解上图中Canal的工作原理需要首先要知道MySQL主从数据同步的原理，如下图：

MySQL主从数据同步的原理

MySQL主从集群由MySQL主服务器(master)和MySQL从服务器(slave)组成，MySQL主从数据同步是一种数据库复制技术，进行写数据会先向主服务器写，写成功后将数据同步到从服务器，流程如下：

1、主服务器将所有写操作（INSERT、UPDATE、DELETE）以二进制日志（binlog）的形式记录下来。

2、从服务器连接到主服务器，发送dump 协议,请求获取主服务器上的binlog日志。

MySQL的dump协议是MySQL复制协议中的一部分。

3、MySQL master 收到 dump 请求，开始推送 binary log 给 slave

4、从服务器解析日志，根据日志内容更新从服务器的数据库，完成从服务器的数据保持与主服务器同步。

理解了MySQL主从同步的原理，Canal在整个过程充当什么角色呢？

如下图：

工作流程如下：

1、Canal模拟 MySQL slave 的交互协议，伪装自己为 MySQL slave ，向 MySQL master 发送dump 协议

MySQL的dump协议是MySQL复制协议中的一部分。

2、MySQL master 收到 dump 请求，开始推送 binary log 给 slave (即 canal )

。一旦连接建立成功，Canal会一直等待并监听来自MySQL主服务器的binlog事件流，当有新的数据库变更发生时MySQL master主服务器发送binlog事件流给Canal。

3、Canal会及时接收并解析这些变更事件并解析 binary log

通过以上流程可知Canal和MySQL master主服务器之间建立了长连接。

流程：运营端人员修改信息，canal监听到mysql的binlog日志并进行解析，发送到mq,mq监听到消息，同步简历索引

MQ技术方案

目标：

能说出如何保证MQ消息的可靠性？

1）保证生产消息可靠性

上述技术方案中有一个关键点，首先数据增删改的信息是保证写入binlog的，Canal解析出增删改的信息后写入MQ，同步程序从MQ去读取消息，如果MQ中的消息丢失了数据将无法进行同步。

如何保证MQ消息的可靠性？

保证MQ消息的可靠性分两个方面：保证生产消息的可靠性、保证消费消息的可靠性。

保证生产消息的可靠性：

RabbitMQ提供生产者确认机制保证生产消息的可靠性，技术方案如下

首先发送消息的方法如果执行失败会进行重试，重试次数耗尽记录失败消息

如果重试失败，则将数据保存到数据库中

通过MQ的提供的生产者确认机制保证生产消息的可靠性

使用生产者确认机制需要给每个消息指定一个唯一ID，生产者确认机制通过异步回调的方式进行，包括ConfirmCallback和Return回调。

ConfirmCallback：消息发送到Broker会有一个结果返回给发送者表示消息是否处理成功：

1）消息成功投递到交换机，返回ack

2）消息未投递到交换机，返回nack

在发送消息时指定回调对象

回调类中回调方法源代码：

如果没有返回ack则将消息记录到失败消息表，如果经过重试后返回了ack说明消息发送成功，此时将消息从失败消息表删除。

Return回调：如果消息发送到交换机成功了但是并没有到达队列，此时会调用ReturnCallback回调方法，在回调方法中我们可以收到失败的消息存入失败消息表以便进行补偿。

要使用Return回调需要开启设置：

首先在shared-rabbitmq.yaml中配置rabbitMQ参数，如下：

spring:rabbitmq:publisher-confirm-type: correlatedpublisher-returns: truetemplate:mandatory: true说明：
publish-confirm-type：开启publisher-confirm，这里支持两种类型：
simple：同步等待confirm结果，直到超时
correlated：异步回调，定义ConfirmCallback，MQ返回结果时会回调这个ConfirmCallback
publish-returns：开启publish-return功能，同样是基于callback机制，不过是定义ReturnCallback
template.mandatory：定义消息路由失败时的策略。true，则调用ReturnCallback；false：则直接丢弃消息

保证消费消息可靠性

首先设置消息持久化，保证消息发送到MQ消息不丢失。具体需要设置交换机和队列支持持久化，发送消息设置deliveryMode=2。

RabbitMQ是通过消费者回执来确认消费者是否成功处理消息的：消费者获取消息后，应该向RabbitMQ发送ACK回执，表明自己已经处理完成消息，RabbitMQ收到ACK后删除消息。

spring:rabbitmq:....listener:simple:acknowledge-mode: auto #，出现异常时返回nack，消息回滚到mq；没有异常，返回ackretry:enabled: true # 开启消费者失败重试initial-interval: 1000 # 初识的失败等待时长为1秒multiplier: 10 # 失败的等待时长倍数，下次等待时长 = multiplier * last-intervalmax-attempts: 3 # 最大重试次数stateless: true # true无状态；false有状态。如果业务中包含事务，这里改为false

能说出如何保证MQ幂等性？或如何防止重复消费？

消费者在消费消息时难免出现重复消费的情况，比如：消费者没有向MQ返回ack导致重复消费，所以消费者需要保证消费消息幂等性。

什么是幂等性？

幂等性是指不论执行多少次其结果是一致的。

举例：

收到消息需要向数据新增一条记录，如果重复消费则会出现重复添加记录的问题。

下边根据场景分析解决方案：

1、查询操作

本身具有幂等性。

2、添加操作

如果主键是自增则可能重复添加记录。

保证幂等性可以设置数据库的唯一约束，比如：添加学生信息，将学号字段设置为唯一索引，即使重复添加相同的学生同一个学号只会添加一条记录。

3、更新操作

如果是更新一个固定的值，比如: update users set status =1 where id=?，本身具有幂等性。

如果只允许更新成功一次则可以使用token机制，发送消息前生成一个token写入redis，收到消息后解析出token从redis查询token如果成功则说明没有消费，此时更新成功将token从redis删除，当重复消费相同的消息时由于token已经从redis删除不会再执行更新操作。

4、删除操作

与更新操作类似，如果是删除某个具体的记录，比如：delete from users where id=?，本身具有幂等性。

如果只允许删除成功一次可以采用更新操作相同的方法。

可以百分百保证MQ的消息可靠性吗？

保证消息可靠性分两个方面：保证生产消息可靠性和保证消费消息可靠性。

保证生产消息可靠性：

生产消息可靠性是通过判断MQ是否发送ack回执，如果发nack表示发送消息失败，此时会进行重发或记录到失败消息表，通过定时任务进行补偿发送。如果Java程序并没有收到回执（如jvm进程异常结束了，或断电等因素），此时将无法保证生产消息的可靠性。

保证消费消息可靠性：

保证消费消息可靠性方案首先保证发送消息设置为持久化，其次通过MQ的消费确认机制保证消费者消费成功消息后再将消息删除。

虽然设置了消息持久化，消息进入MQ首先是在缓存存在，MQ会根据一定的规则进行刷盘，比如：每隔几毫秒进行刷盘，如果在消息还没有保存到磁盘时MQ进程终止，此时将会丢失消息。虽然可以使用镜像队列（用于在 RabbitMQ 集群中复制队列的消息，这样做的目的是提高队列的可用性和容错性，以防止在单个节点故障时导致消息的丢失。）但也不能百分百保证消息不丢失。

虽然我们加了很多保证可靠性的机制，这样也只是去提高消息的可靠性，不能百分百做的可靠，所以使用MQ的场景要考虑这种问题的存在，做好补偿处理任务。

Canal+MQ同步流程

实现将MySQL的变更数据通过Canal写入MQ。

配置Canal+MQ数据同步环境

参考配置链接

Docshttps://mx67xggunk5.feishu.cn/wiki/Yifpw51Qoim81akHF7ic5ye3npd

根据Canal+MQ同步流程，下边进行如下配置：

配置Mysql主从同步，开启MySQL主服务器的binlog
安装Canal并配置，保证Canal连接MySQL主服务器成功
安装RabbitMQ，并配置同步队列。
在Canal中配置RabbitMQ的连接信息，保证Canal收到binlog消息写入MQ

索引同步

因为涉及到好几张表，为了降低代码复杂度，把好几张表中的字段都聚合到一张表中，那么canal只需要监听这一张表的数据就可以

上面通过配置Canal+MQ的数据同步环境实现了Canal从数据库读取binlog并且将数据写入MQ。

下边编写同步程序监听MQ，收到消息后向ES创建索引。

es的版本是7.17.7，不同的版本api不一样

keyword是关键词，表示不对该属性进行分词，index:false表示不对其进行索引，还有分词方式

PUT /serve_aggregation
{"mappings" : {"properties" : {"city_code" : {"type" : "keyword"},"detail_img" : {"type" : "text","index" : false},"hot_time_stamp" : {"type" : "long"},"id" : {"type" : "keyword"},"is_hot" : {"type" : "short"},"price" : {"type" : "double"},"serve_item_icon" : {"type" : "text","index" : false},"serve_item_id" : {"type" : "keyword"},"serve_item_img" : {"type" : "text","index" : false},"serve_item_name" : {"type" : "text","analyzer": "ik_max_word","search_analyzer":"ik_smart"},"serve_item_sort_num" : {"type" : "short"},"serve_type_icon" : {"type" : "text","index" : false},"serve_type_id" : {"type" : "keyword"},"serve_type_img" : {"type" : "text","index" : false},"serve_type_name" : {"type" : "text","analyzer": "ik_max_word","search_analyzer":"ik_smart"},"serve_type_sort_num" : {"type" : "short"}}}
}

编写监听器监听canal的数据变化

concurrency = "1"：表示消费线程数为1。

在同步程序中需要根据业务需求编写同步方法，当服务下架时会删除索引需要重写抽象类中的batchDelete(List<Long> ids)方法，此方法是当删除Serve_sync表的记录时对索引执行删除操作。

当服务上架后需要添加索引，当服务信息修改时需要修改索引，需要重写抽象类中的batchSave(List<ServeSync> data)方法，此方法是当向Serve_sync表新增或修改记录时对索引执行添加及修改操作。

mq监听到消息后解析消息

/*** 服务信息同步程序** @author itcast* @create 2023/8/15 18:14**/
@Component
public class ServeCanalDataSyncHandler extends AbstractCanalRabbitMqMsgListener<ServeSync> {@Resourceprivate ElasticSearchTemplate elasticSearchTemplate;@RabbitListener(bindings = @QueueBinding(value = @Queue(name = "canal-mq-jzo2o-foundations"),exchange = @Exchange(name = "exchange.canal-jzo2o", type = ExchangeTypes.TOPIC),key = "canal-mq-jzo2o-foundations"),concurrency = "1")public void onMessage(Message message) throws Exception {parseMsg(message);}@Overridepublic void batchSave(List<ServeSync> data) {Boolean aBoolean = elasticSearchTemplate.opsForDoc().batchInsert(IndexConstants.SERVE, data);if(!aBoolean){try {Thread.sleep(1000);} catch (InterruptedException e) {throw new RuntimeException(e);}throw new RuntimeException("同步失败");}}@Overridepublic void batchDelete(List<Long> ids) {Boolean aBoolean = elasticSearchTemplate.opsForDoc().batchDelete(IndexConstants.SERVE, ids);if(!aBoolean){try {Thread.sleep(1000);} catch (InterruptedException e) {throw new RuntimeException(e);}throw new RuntimeException("同步失败");}}
}

启动jzo2o-foundations服务。

启动成功,jzo2o-foundations服务作为MQ的消费者和MQ建立通道，进入canal-mq-jzo2o-foundations队列的管理界面，查看是否建立了监听通道。

解析完数据之后判断是否是读多个数据，多个数据就执行批量方法，

然后判断是否是保存新增或者删除。

此时运行程序，看是否能正常接收到消息。

如何保证Canal+MQ同步消息的顺序性？

场景：

如下图：

首先明确Canal解析binlog日志信息按顺序发到MQ的队列中，现在是要保证消费端如何按顺序消费队列中的消息。

生产中同一个jzo2o-foundations服务会启动多个jvm进程，每个进程作为canal-mq-jzo2o-foundations的消费者,如下图：

    @RabbitListener(bindings = @QueueBinding(value = @Queue(name = "canal-mq-jzo2o-foundations",arguments={@Argument(name="x-single-active-consumer", value = "true", type = "java.lang.Boolean") }),exchange = @Exchange(name="exchange.canal-jzo2o",type = ExchangeTypes.TOPIC),key="canal-mq-jzo2o-foundations"),concurrency="1")public void onMessage(Message message) throws Exception{parseMsg(message);}

管理同步表

通过测试Canal+MQ同步流程，只有当serve_sync表变化时才会触发同步，serve_sync表什么时候变化？

/*** 新增服务同步数据** @param serveId 服务id*/
private void addServeSync(Long serveId) {//服务信息Serve serve = baseMapper.selectById(serveId);//区域信息Region region = regionMapper.selectById(serve.getRegionId());//服务项信息ServeItem serveItem = serveItemMapper.selectById(serve.getServeItemId());//服务类型ServeType serveType = serveTypeMapper.selectById(serveItem.getServeTypeId());ServeSync serveSync = new ServeSync();serveSync.setServeTypeId(serveType.getId());serveSync.setServeTypeName(serveType.getName());serveSync.setServeTypeIcon(serveType.getServeTypeIcon());serveSync.setServeTypeImg(serveType.getImg());serveSync.setServeTypeSortNum(serveType.getSortNum());serveSync.setServeItemId(serveItem.getId());serveSync.setServeItemIcon(serveItem.getServeItemIcon());serveSync.setServeItemName(serveItem.getName());serveSync.setServeItemImg(serveItem.getImg());serveSync.setServeItemSortNum(serveItem.getSortNum());serveSync.setUnit(serveItem.getUnit());serveSync.setDetailImg(serveItem.getDetailImg());serveSync.setPrice(serve.getPrice());serveSync.setCityCode(region.getCityCode());serveSync.setId(serve.getId());serveSync.setIsHot(serve.getIsHot());serveSyncMapper.insert(serveSync);
}

省略修改服务类型，服务项时更改 sync表的代码

搜索接口

目标：开发搜索接口。

@RestController("consumerServeController")
@RequestMapping("/customer/serve")
@Api(tags = "用户端 - 首页服务查询接口")
public class FirstPageServeController {
...@GetMapping("/search")@ApiOperation("首页服务搜索")@ApiImplicitParams({@ApiImplicitParam(name = "cityCode", value = "城市编码", required = true, dataTypeClass = String.class),@ApiImplicitParam(name = "serveTypeId", value = "服务类型id", dataTypeClass = Long.class),@ApiImplicitParam(name = "keyword", value = "关键词", dataTypeClass = String.class)})public List<ServeSimpleResDTO> findServeList(@RequestParam("cityCode") String cityCode,@RequestParam(value = "serveTypeId", required = false) Long serveTypeId,@RequestParam(value = "keyword", required = false) String keyword) {return null;}

首先通过ES的查询语言进行查询，如下：

GET /serve_aggregation/_search
{"query" : {"bool" : {"must" : [{"term" : {"city_code" : {"value" : "010"}}},{"multi_match" : {"fields" : [ "serve_item_name", "serve_type_name" ],"query" : "保洁"}}]}},"sort" : [{"serve_item_sort_num" : {"order" : "asc"}}]
}