【解析 RocketMQ 业务消息--“顺序消息”】简介： 本篇将继续业务消息集成的场景，从功能原理、应用案例、最佳实践以及实战等角度介绍 RocketMQ 的顺序消息功能。作者：绍舒 引言 Apache RocketMQ 诞生至今，历经十余年大规模业务稳定性打磨，服务了阿里集团内部业务以及阿里云数以万计的企业客户。作为金融级可靠的业务消息方案，RocketMQ 从创建之初就一直专注于业务集成领域的异步通信能力构建。本篇将继续业务消息集成的场景，从功能原理、应用案例、最佳实践以及实战等角度介绍 RocketMQ 的顺序消息功能。 简介 顺序消息是消息队列 RocketMQ 版提供的一种对消息发送和消费顺序有严格要求的消息。对于一个指定的 Topic，同一 MessageGroup 的消息按照严格的先进先出（FIFO）原则进行发布和消费，即先发布的消息先消费，后发布的消息后消费，服务端严格按照发送顺序进行存储、消费。同一 MessageGroup 的消息保证顺序，不同 MessageGroup 之间的消息顺序不做要求，因此需做到两点，发送的顺序性和消费的顺序性。

文章图片

功能原理 在这里首先抛出一个问题，在日常的接触中，许多 RocketMQ 使用者会认为，既然顺序消息能在普通消息的基础上实现顺序，看起来就是普通消息的加强版，那么为什么不全部都使用顺序消息呢？接下来就会围绕这个问题，对比普通消息和顺序消息进行阐述。 顺序发送 在分布式环境下，保证消息的全局顺序性是十分困难的，例如两个 RocketMQ Producer A 与 Producer B，它们在没有沟通的情况下各自向 RocketMQ 服务端发送消息 a 和消息 b，由于分布式系统的限制，我们无法保证 a 和 b 的顺序。因此业界消息系统通常保证的是分区的顺序性，即保证带有同一属性的消息的顺序，我们将该属性称之为 MessageGroup。如图所示，ProducerA 发送了 MessageGroup 属性为 A 的两条消息 A1，A2 和 MessageGroup 属性为 B 的 B1，B2，而 ProducerB 发送了 MessageGroup 属性为 C 的两条属性 C1，C2。

文章图片

同时，对于同一 MessageGroup，为了保证其发送顺序的先后性，比较简单的做法是构造一个单线程的场景，即不同的 MessageGroup 由不同的 Producer 负责，并且对于每一个 Producer 而言，顺序消息是同步发送的。同步发送的好处是显而易见的，在客户端得到上一条消息的发送结果后再发送下一条，即能准确保证发送顺序，若使用异步发送或多线程则很难保证这一点。

文章图片

因此可以看到，虽然在底层原理上，顺序消息发送和普通消息发送并无二异，但是为了保证顺序消息的发送顺序性，同步发送的方式相比较普通消息，实际上降低了消息的最大吞吐。 顺序消费 与顺序消息不同的是，普通消息的消费实际上没有任何限制，消费者拉取的消息是被异步、并发消费的，而顺序消息，需要保证对于同一个 MessageGroup，同一时刻只有一个客户端在消费消息，并且在该条消息被确认消费完成之前（或者进入死信队列），消费者无法消费同一 MessageGroup 的下一条消息，否则消费的顺序性将得不到保证。因此这里存在着一个消费瓶颈，该瓶颈取决于用户自身的业务处理逻辑。极端情况下当某一 MessageGroup 的消息过多时，就可能导致消费堆积。当然也需要明确的是，这里的语境都指的是同一 MessageGroup，不同 MessageGroup 的消息之间并不存在顺序性的关联，是可以进行并发消费的。因此全文中提到的顺序实际上是一种偏序。

文章图片

小结 无论对于发送还是消费，我们通过 MessageGroup 的方式将消息分组，即并发的基本单元是 MessageGroup，不同的 MessageGroup 可以并发的发送和消费，从而一定程度具备了可拓展性，支持多队列存储、水平拆分、并发消费，且不受影响。回顾普通消息，站在顺序消息的视角，可以认为普通消息的并发基本单元是单条消息，即每条消息均拥有不同的 MessageGroup。 我们回到开头那个问题： 既然顺序消息能在普通消息的基础上实现顺序，看起来就是普通消息的加强版，那么为什么不全部都使用顺序消息呢？ 现在大家对于这个问题可能有一个基本的印象了，消息的顺序性当然很好，但是为了实现顺序性也是有代价的。 下述是一个表格，简要对比了顺序消息和普通消息。

文章图片

最佳实践 合理设置 MessageGroup MessageGroup 会有很多错误的选择，以某电商平台为例，某电商平台将商家 ID 作为 MessageGroup，因为部分规模较大的商家会产出较多订单，由于下游消费能力的限制，因此这部分商家所对应的订单就发生了严重的堆积。正确的做法应当是将订单号作为 MessageGroup，而且站在背后的业务逻辑上来说，同一订单才有顺序性的要求。即选择 MessageGroup 的最佳实践是：MessageGroup 生命周期最好较为短暂，且不同 MessageGroup 的数量应当尽量相同且均匀。 同步发送和发送重试 如之前章节所述，需使用同步发送和发送重试来保证发送的顺序性。 消费幂等 消息传输链路在异常场景下会有少量重复，业务消费是需要做消费幂等，避免重复处理带来的风险。 应用案例 用户注册需要发送验证码，以用户 ID 作为 MessageGroup，那么同一个用户发送的消息都会按照发布的先后顺序来消费。 电商的订单创建，以订单 ID 作为 MessageGroup，那么同一个订单相关的创建订单消息、订单支付消息、订单退款消息、订单物流消息都会按照发布的先后顺序来消费。

文章图片

实战 发送 可以看到，该发送案例设置了 MessageGroup 并且使用了同步发送，发送的代码如下： public class ProducerFifoMessageExample {

private static final Logger LOGGER = LoggerFactory.getLogger(ProducerFifoMessageExample.class); private ProducerFifoMessageExample() { }public static void main(String[] args) throws ClientException, IOException { final ClientServiceProvider provider = ClientServiceProvider.loadService(); // Credential provider is optional for client configuration. String accessKey = "yourAccessKey"; String secretKey = "yourSecretKey"; SessionCredentialsProvider sessionCredentialsProvider = new StaticSessionCredentialsProvider(accessKey, secretKey); String endpoints = "foobar.com:8080"; ClientConfiguration clientConfiguration = ClientConfiguration.newBuilder() .setEndpoints(endpoints) .setCredentialProvider(sessionCredentialsProvider) .build(); String topic = "yourFifoTopic"; final Producer producer = provider.newProducerBuilder() .setClientConfiguration(clientConfiguration) // Set the topic name(s), which is optional. It makes producer could prefetch the topic route before // message publishing. .setTopics(topic) // May throw {@link ClientException} if the producer is not initialized. .build(); // Define your message body. byte[] body = "This is a FIFO message for Apache RocketMQ".getBytes(StandardCharsets.UTF_8); String tag = "yourMessageTagA"; final Message message = provider.newMessageBuilder() // Set topic for the current message. .setTopic(topic) // Message secondary classifier of message besides topic. .setTag(tag) // Key(s) of the message, another way to mark message besides message id. .setKeys("yourMessageKey-1ff69ada8e0e") // Message group decides the message delivery order. .setMessageGroup("youMessageGroup0") .setBody(body) .build(); try { final SendReceipt sendReceipt = producer.send(message); LOGGER.info("Send message successfully, messageId={}", sendReceipt.getMessageId()); } catch (Throwable t) { LOGGER.error("Failed to send message", t); } // Close the producer when you don't need it anymore. producer.close(); }

} 消费 消费的代码如下： public class SimpleConsumerExample {

private static final Logger LOGGER = LoggerFactory.getLogger(SimpleConsumerExample.class); private SimpleConsumerExample() { }public static void main(String[] args) throws ClientException, IOException { final ClientServiceProvider provider = ClientServiceProvider.loadService(); // Credential provider is optional for client configuration. String accessKey = "yourAccessKey"; String secretKey = "yourSecretKey"; SessionCredentialsProvider sessionCredentialsProvider = new StaticSessionCredentialsProvider(accessKey, secretKey); String endpoints = "foobar.com:8080"; ClientConfiguration clientConfiguration = ClientConfiguration.newBuilder() .setEndpoints(endpoints) .setCredentialProvider(sessionCredentialsProvider) .build(); String consumerGroup = "yourConsumerGroup"; Duration awaitDuration = Duration.ofSeconds(30); String tag = "yourMessageTagA"; String topic = "yourTopic"; FilterExpression filterExpression = new FilterExpression(tag, FilterExpressionType.TAG); SimpleConsumer consumer = provider.newSimpleConsumerBuilder() .setClientConfiguration(clientConfiguration) // Set the consumer group name. .setConsumerGroup(consumerGroup) // set await duration for long-polling. .setAwaitDuration(awaitDuration) // Set the subscription for the consumer. .setSubscriptionExpressions(Collections.singletonMap(topic, filterExpression)) .build(); // Max message num for each long polling. int maxMessageNum = 16; // Set message invisible duration after it is received. Duration invisibleDuration = Duration.ofSeconds(5); final List messages = consumer.receive(maxMessageNum, invisibleDuration); for (MessageView message : messages) { try { consumer.ack(message); } catch (Throwable t) { LOGGER.error("Failed to acknowledge message, messageId={}", message.getMessageId(), t); } } // Close the simple consumer when you don't need it anymore. consumer.close(); }

} 今天通过对 RocketMQ 顺序消息的介绍，希望能够帮大家对顺序消息的原理和应用有更深入的了解，同时也期望 RocketMQ 的顺序消息能够帮助您更有效的解决业务问题。如果您对 RocktMQ 的业务消息感兴趣，也欢迎您扫描下方二维码加入钉钉群一起沟通交流～

文章图片

点击此处，进入官网了解更多详情～原文链接：https://click.aliyun.com/m/10...本文为阿里云原创内容，未经允许不得转载。

推荐阅读

• 

  智慧保险的快速发展,未来的保险公司将是什么模样呢？
• 

  获取oss的pdf文件，返回文件输出流
• 

  2023重庆涪陵3月赏花时间+地点 涪陵的花
• 

  wnda是什么轮胎 wnda是什么轮胎里面有一层胶
• 

  时光与岁月优美短句 关于时光岁月的唯美句子
• 

  mysql怎么设置字段默认值为0 mysql设置字段为小时
• 

  魔兽世界桃乐豆详细介绍
• 

  食色,性也究竟是什么意思 食色性也是什么意思
• 

  什么材质的行李箱好一点 什么材质的行李箱好
• 

  华为网盘会被和谐吗,华为网盘有什么用？
• 

  成都徒步旅行～懒鱼第四天作业
• 

  关系胜于一切，关系就是温暖
• 

  g三星 Galaxy S22 / S22 +图赏：老设计焕发新感觉
• 

  北京医保提到4000北京医保提到4000的原因（北京医保提到4000北京医保
• 

  文字代码生成器，谁有日志彩字生成代码器
• 

  在网店转让平台上如何让自己的店铺快速出售？有哪些小技巧？
• 

  十七岁，我曾有一个朋友
• 

  潇潇雨歇是什么意思 潇潇雨歇解释
• 

  四轴数控车床怎么编程，请大师教教我这个用四轴怎么编程
• 

  怎么判断水果是否有毒

• 一文搞懂 SAE 日志采集架构
• 基于 Serverless+OSS 分分钟实现图片秒变素描
• springboot|一文详解Spring、SpringBoot、Springcloud的关系与区别
• #|16-Spring AOP源码分析-@EnableAspectJAutoProxy和AspectJAutoProxyRegistrar
• 动态尺寸模型优化实践之Shape Constraint IR Part II
• 动态尺寸模型优化实践之Shape Constraint IR Part I
• 782. 变为棋盘 : 构造分析题
• 后端|一个忙碌架构师的Java后端书架（2022）
• java|2.5 万字详解（23 种设计模式,阿里京东面试必过）