如何实现保证Canal+MQ消息同步顺序性

文
章
目
录

涉及数据同步的项目时，经常会遇到需要保证消息顺序性的场景。就像在通过Elasticsearch实现服务搜索功能的过程中，会用到Canal+MQ来完成服务信息与ES索引的同步。在这个同步过程里，有一个关键问题，那就是如何确保消息顺序性。接下来，咱们详细探讨一下。

一、明确需求

在使用Canal+MQ进行服务信息与ES索引同步时，Canal负责解析binlog日志信息，并将这些信息发送到MQ的队列中。当前的重点在于，要保证消费端能按照正确的顺序消费队列中的消息。在实际生产环境中，同一个jzo2o-foundations服务可能会启动多个JVM进程，每个进程都作为canal-mq-jzo2o-foundations的消费者。

想象一下，多个JVM进程就好比多个“收件人”，都在等待从同一个“邮箱”（队列）中收取消息。如果这些“收件人”同时去取消息，而且没有特定的顺序规则，那么最终收到消息的顺序就可能混乱，这会导致处理结果和我们预期的不一样。比如，某些数据的更新操作，如果顺序乱了，可能会使数据处于错误的状态，影响整个系统的正常运行。

二、寻找解决方法

为了解决消息顺序混乱的问题，我们可以采用消费队列中的数据使用单线程的方式。简单来说，就是让多个JVM进程在监听同一个队列时，保证只有一个消费者处于活跃状态，这样就能确保只有一个“收件人”在同一时间从“邮箱”取消息，从而控制消息的消费顺序。

三、具体实现：保证只有一个消费者接收消息

要实现只有一个消费者接收消息，可以在队列中增加x-single-active-consumer参数，这个参数的作用是启用单一活动消费者模式。

在创建队列时，相关的配置操作如下：

队列配置：在创建队列时，指定虚拟主机为/xzb，队列类型选择Classic ，名称设置为canal-mq-jzo20-foundation ，设置队列持久化（Durability为Durable），不自动删除（Auto delete为No），并在Arguments中添加x-single-active-consumer true 。完成配置后，可以查看队列，确保队列上存在SAC标识。例如，在查看队列信息时，如果看到类似/xzb | canal-mg-jzo2o-foundations | cassic | D SAC | Args | idle这样的记录，就说明配置生效了。
当有多个jvm进程都去监听该队列时，只有一个为活跃状态
代码配置：如果在代码中使用这个配置，以Java代码为例，在@RabbitListener注解中进行如下设置：

// 在@RabbitListener注解中，配置队列、交换机、路由键以及消费线程数等信息
@RabbitListener(bindings = @QueueBinding(
        // 配置队列，设置队列名为canal-mq-jzo2o-foundations，并添加x-single-active-consumer参数
        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"),
        // 指定消费线程为1，确保单线程消费消息
        concurrency="1"
)
// 定义消息处理方法，处理接收到的消息
public void onMessage(Message message) throws Exception{
    parseMsg(message);
}

在上述代码中，arguments={@Argument(name="x-single-active-consumer", value = "true", type = "java.lang.Boolean") }用于设置队列的x-single-active-consumer参数为true ，开启单一活动消费者模式；concurrency="1"表示指定消费线程为1，保证单线程消费消息，从而确保消息按顺序被处理。