消息队列 RocketMQ版
消息队列 RocketMQ版
文档指南
请输入
- 文档首页
消息队列 RocketMQ版SDK 参考RocketMQ 5.xJava 5.x SDK顺序消息
复制全文
下载 pdf顺序消息
文档反馈
问问助手消息队列 RocketMQ 5.x 版本提供顺序消息（FIFO 消息）供您使用。在顺序消息模型中，您需要严格按照顺序来发布和消费消息。本文提供使用 Java SDK 收发顺序消息的示例代码供您参考。
背景信息
- 使用顺序消息前，需在控制台创建顺序消息类型的 Topic 和 Group。
- 顺序消息分为两类，全局顺序消息和分区顺序消息。区别仅为队列数量不同，代码没有区别。
- 全局顺序：
- 对于指定的一个 Topic，所有消息的生产和消费需要遵循一定的顺序，消息的消费顺序必须和生产顺序一致，即需要严格的先入先出 FIFO（First In First Out）的顺序进行发布和消费。
- 分区顺序：
- 对于指定的一个 Topic，其中每一个分区的消息生产与消费是有序的，同一个队列内的消息按照严格的 FIFO 顺序进行发布和订阅。消息投递到哪一个分区由消息的messageGroup来进行区分。
发送顺序消息
发送顺序消息的示例代码如下。
public class RocketMQOrderProducer {
private static final Logger log = LoggerFactory.getLogger(RocketMQOrderProducer.class);
private RocketMQOrderProducer() {
}
public static void main(String[] args) throws ClientException, IOException {
final ClientServiceProvider provider = ClientServiceProvider.loadService();
// 设置为您从火山引擎消息队列 RocketMQ版控制台获取的接入点信息，类似“{INSTANCE_ID}.rocketmq.ivolces.com:8080”。
String endpoints = "rocketmq-xxx.rocketmq.ivolces.com:8080";
// 根据实例支持的认证方式，填写对应的身份认证信息：
// 密钥管理：accessKey 为 AccessKey ID，secretKey 为 AccessKey Secret。
// 权限管理：accessKey 为 ACL 用户名，secretKey 为 ACL 密码。
String accessKey = "xxxx";
String secretKey = "xxxx";
SessionCredentialsProvider sessionCredentialsProvider =
new StaticSessionCredentialsProvider(accessKey, secretKey);
ClientConfiguration clientConfiguration = ClientConfiguration.newBuilder()
.setEndpoints(endpoints)
// 暂不支持 SSL 模式。
.enableSsl(false)
.setCredentialProvider(sessionCredentialsProvider)
.build();
String topic = "yourTopicName"; // 已在控制台创建的顺序消息类型 Topic。
final Producer producer = provider.newProducerBuilder()
.setClientConfiguration(clientConfiguration)
.setTopics(topic)
.build();
// 定义消息体。
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("yourMessageGroup0")
.setBody(body)
.build();
try {
final SendReceipt sendReceipt = producer.send(message);
log.info("Send message successfully, messageId={}", sendReceipt.getMessageId());
} catch (Throwable t) {
log.error("Failed to send message", t);
}
// Close the producer when you don't need it anymore.
// You could close it manually or add this into the JVM shutdown hook.
producer.close();
}
}
订阅顺序消息
订阅顺序消息的示例代码如下。使用前请确认所用 Group 为顺序消息类型。
public class RocketMQOrderConsumer {
private static final Logger log = LoggerFactory.getLogger(RocketMQOrderConsumer.class);
private RocketMQOrderConsumer() {
}
public static void main(String[] args) throws ClientException, InterruptedException, IOException {
final ClientServiceProvider provider = ClientServiceProvider.loadService();
// 设置为您从火山引擎消息队列 RocketMQ版控制台获取的接入点信息，类似“{INSTANCE_ID}.rocketmq.ivolces.com:8080”。
String endpoints = "rocketmq-xxx.rocketmq.ivolces.com:8080";
// 根据实例支持的认证方式，填写对应的身份认证信息：
// 密钥管理：accessKey 为 AccessKey ID，secretKey 为 AccessKey Secret。
// 权限管理：accessKey 为 ACL 用户名，secretKey 为 ACL 密码。
String accessKey = "xxxx";
String secretKey = "xxxx";
SessionCredentialsProvider sessionCredentialsProvider =
new StaticSessionCredentialsProvider(accessKey, secretKey);
ClientConfiguration clientConfiguration = ClientConfiguration.newBuilder()
.setEndpoints(endpoints)
// 暂不支持 SSL 模式。
.enableSsl(false)
.setCredentialProvider(sessionCredentialsProvider)
.build();
String tag = "yourMessageTagA";
FilterExpression filterExpression = new FilterExpression(tag, FilterExpressionType.TAG);
String consumerGroup = "yourConsumerGroup"; // 已在控制台创建的顺序消息类型 Group。
String topic = "yourTopic"; // 已在控制台创建的顺序消息类型 Topic。
// In most case, you don't need to create too many consumers, singleton pattern is recommended.
PushConsumer pushConsumer = provider.newPushConsumerBuilder()
.setClientConfiguration(clientConfiguration)
// Set the consumer group name.
.setConsumerGroup(consumerGroup)
// Set the subscription for the consumer.
.setSubscriptionExpressions(Collections.singletonMap(topic, filterExpression))
.setMessageListener(messageView -> {
// Handle the received message and return consume result.
log.info("Consume message={}", messageView);
return ConsumeResult.SUCCESS;
})
.build();
// Block the main thread, no need for production environment.
Thread.sleep(Long.MAX_VALUE);
// Close the push consumer when you don't need it anymore.
// You could close it manually or add this into the JVM shutdown hook.
pushConsumer.close();
}
}
最近更新时间：2026.08.26 15:19:55
这个页面对您有帮助吗？
有用
有用
无用
无用