RocketMQ保证消息的有序性的案例分享
在分布式系统中，消息队列（MQ）的有序性是一个重要的特性，尤其是在需要保证事件顺序执行的业务场景下。Apache RocketMQ 是一个常用的开源消息中间件，它提供了强大的有序消息处理能力。这里我们会探讨 RocketMQ 是如何保证消息的有序性的，包括其设计原理和相关的源码实现。
简单原理
RocketMQ 有序消息的基本概念
RocketMQ 保证有序性的主要方法是通过顺序消息来实现的。在 RocketMQ 中，顺序消息分为全局顺序和分区顺序两种：
- 全局顺序：指的是消息全局范围内的有序，也就是在所有的消息中，都是按照发送的顺序来消费。
- 分区顺序：指的是在同一个队列（Queue）中的消息是有序的，而不同队列间的消息并不保证有序。
RocketMQ 默认使用分区顺序，通过将同一个 topic 下的消息分到同一个队列（queue）中，来保证队列内的消息有序。
RocketMQ 有序消息的实现机制
消息发送
在发送端，RocketMQ 通过确保生产者向同一个队列（Queue）发送消息来保证消息的有序性。生产者在发送消息时可以指定消息的 keys 或者其他属性，RocketMQ 通过这些属性计算消息应该发送到哪个队列。
源码示例（伪代码）：
public class Producer {
public void sendMessages(List<Message> messages) {
for (Message msg : messages) {
int queueId = this.calculateQueueId(msg);
msg.setQueueId(queueId);
this.sendMessageToQueue(msg, queueId);
}
}
private int calculateQueueId(Message msg) {
// 使用 hash 算法基于 message key 计算队列 ID
return Math.abs(msg.getKey().hashCode()) % this.queueSize;
}
}
消息消费
在消费端，RocketMQ 使用单线程消费模式来保证同一个队列的消息顺序性。消费者会固定分配到某个队列，而且是单线程从该队列拉取并处理消息，从而保证消息的有序处理。
源码示例（伪代码）：
public class Consumer {
public void consume() {
while (true) {
Message msg = this.pullMessage();
this.processMessage(msg);
}
}
}
简单案例
在Spring Boot中使用RocketMQ来保证消息的队列顺序性，我们需要配置RocketMQ的客户端和服务器端以支持顺序消息。以下是一个基于RocketMQ和Spring Boot实现的消息顺序发送和消费的例子。这个场景假设我们需要在一个电商系统中处理订单状态更新，订单状态更新必须按照顺序来处理，以避免状态不一致。
步骤1: 添加依赖
首先，确保你的pom.xml中加入了RocketMQ的Spring Boot Starter依赖。
<dependencies>
<dependency>
<groupId>org.apache.rocketmq</groupId>
<artifactId>rocketmq-spring-boot-starter</artifactId>
<version>2.1.0</version>
</dependency>
</dependencies>
步骤2: 配置RocketMQ
在application.yml或application.properties中配置RocketMQ的基础属性：
rocketmq:
name-server: 127.0.0.1:9876 # 修改为你的NameServer地址
producer:
group: order_producer_group
send-message-timeout: 3000
consumer:
group: order_consumer_group
consume-thread-min: 1
consume-thread-max: 1
步骤3: 生产者配置
创建一个生产者服务，这个服务将订单状态更新作为顺序消息发送。
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
@Service
public class OrderStatusProducer {
@Autowired
private RocketMQTemplate rocketMQTemplate;
public void sendOrderStatusUpdate(String orderStatus, String orderId) {
// 使用订单ID作为key保证同一个订单的更新在同一个队列
SendResult result = rocketMQTemplate.syncSendOrderly("order-topic", orderStatus, orderId);
System.out.println("Message sent, result: " + result.getSendStatus());
}
}
步骤4: 消费者配置
创建一个消费者服务，这个服务将按顺序消费订单状态更新。
import org.apache.rocketmq.spring.annotation.ConsumeMode;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.springframework.stereotype.Service;
@Service
@RocketMQMessageListener(topic = "order-topic", consumerGroup = "order_consumer_group", consumeMode = ConsumeMode.ORDERLY)
public class OrderStatusConsumer implements RocketMQListener<String> {
@Override
public void onMessage(String message) {
System.out.println("Received order update: " + message);
// 处理订单更新逻辑
processOrderUpdate(message);
}
private void processOrderUpdate(String status) {
// 实现订单更新处理逻辑
System.out.println("Processing order status update: " + status);
}
}
步骤5: 测试消息顺序
你可以通过编写一个简单的测试来发送多个消息，并观察消费者是否按顺序接收它们。
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.CommandLineRunner;
import org.springframework.stereotype.Component;
@Component
public class OrderStatusTestRunner implements CommandLineRunner {
@Autowired
private OrderStatusProducer producer;
@Override
public void run(String... args) throws Exception {
producer.sendOrderStatusUpdate("Order Created", "OrderId123");
producer.sendOrderStatusUpdate("Payment Received", "OrderId123");
producer.sendOrderStatusUpdate("Shipped", "OrderId123");
producer.sendOrderStatusUpdate("Delivered", "OrderId123");
}
}
通过这个设置，RocketMQ 和 Spring Boot 能够保证同一个订单的不同状态更新是按照发送顺序被处理的。这对于需要顺序一致性的业务逻辑是非常重要的。
以上就是RocketMQ保证消息的有序性的案例分享的详细内容，更多关于RocketMQ消息有序性的资料请关注脚本之家其它相关文章！
相关文章
在本文里我们通过实例给大家讲解了JAVA开启线程的方法和相关知识点，需要的朋友们跟着学习下。2019-03-03
这篇文章主要介绍了Mybatis中Mapper映射文件使用详解，文中通过示例代码介绍的非常详细，对大家的学习或者工作具有一定的参考学习价值，需要的朋友们下面随着小编来一起学习学习吧2020-06-06
Eclipse+Java+Swing+Mysql实现电影购票系统(详细代码)
这篇文章主要介绍了Eclipse+Java+Swing+Mysql实现电影购票系统并附详细的代码详解，需要的小伙伴可以参考一下2022-01-01
这篇文章主要介绍了Java的switch为什么不支持long，具有很好的参考价值，希望对大家有所帮助。如有错误或未考虑完全的地方，望不吝赐教2021-10-10
这篇文章主要介绍了Java实现的DES加密解密工具类,结合具体实例形式分析了Java实现的DES加密解密工具类定义与使用方法,需要的朋友可以参考下2017-09-09
这篇文章给大家介绍了新版idea创建spring boot项目的详细教程,本教程对新手小白友好,若根据教程创建出现问题导致失败可下载我提供的源码,在文章最后,本教程较新,文中通过图文给大家介绍的非常详细,感兴趣的朋友可以参考下2024-01-01
这篇文章主要为大家详细介绍了java swing实现电影购票系统，文中示例代码介绍的非常详细，具有一定的参考价值，感兴趣的小伙伴们可以参考一下2019-01-01
详解SpringSecurity中的Authentication信息与登录流程
这篇文章主要介绍了SpringSecurity中的Authentication信息与登录流程,本文给大家介绍的非常详细，对大家的学习或工作具有一定的参考借鉴价值，需要的朋友可以参考下2020-09-09
我们在idea配置maven环境的时候会发现maven更新慢的现象,解决办法就是下载国内的镜像包,完美解决下载速度慢的问题,文中有详细的具体操作方法,并通过图文介绍的非常详细,需要的朋友可以参考下2024-02-02
SpringBoot + Mybatis Plus 整合 Redis的
文章详细介绍了Redis在用户管理系统中的应用,包括用户信息缓存、Token存储、接口限流、重复提交拦截和热点数据预加载等场景,并提供了具体的实现方案和步骤,感兴趣的朋友一起看看吧2025-03-03
最新评论