消息队列 RocketMQ版
消息队列 RocketMQ版
消息队列 RocketMQ版
文档指南
请输入
产品动态
用户指南
实例管理
后台任务管理
Terraform
RocketMQ Agent
API 参考
实例管理
Topic管理
Group管理
消息查询
死信队列
密钥管理
风险巡检
白名单管理
终端节点服务
SDK 参考
RocketMQ 5.x
RocketMQ 4.x
RocketMQ 管控面SDK
最佳实践
- 文档首页
消息队列 RocketMQ版SDK 参考RocketMQ 4.xGo SDK顺序消息
复制全文
下载 pdf顺序消息
文档反馈
问问助手消息队列 RocketMQ版提供顺序消息（FIFO 消息）供您使用。在顺序消息模型中，您需要严格按照顺序来发布和消费消息。本文提供使用 Go SDK 收发顺序消息的示例代码供您参考。
背景信息
顺序消息分为两类，全局顺序消息和分区顺序消息。区别仅为队列数量不同，代码没有区别。
- 全局顺序消息：
- 对于指定的一个 Topic，所有消息的生产和消费需要遵循一定的顺序，消息的消费顺序必须和生产顺序一致，即需要严格的先入先出 FIFO（First In First Out）的顺序进行发布和消费。
- 分区顺序消息：
- 对于指定的一个 Topic，其中每一个分区的消息生产与消费是有序的，同一个队列内的消息按照严格的 FIFO 顺序进行发布和订阅。消息投递到哪一个分区由消息的Sharding Key来进行区分。在 SDK 中可以通过指定 Sharding Key 和 MessageQueueSelector 回调函数来控制消息投递到哪个分区。
前提条件
- 已完成准备工作。
- 已阅读参数说明，了解常用参数的配置方式与填写格式。
发送顺序消息
发送顺序消息的示例代码如下。
import (
"context"
"fmt"
"os"
"strconv"
"github.com/apache/rocketmq-client-go/v2"
"github.com/apache/rocketmq-client-go/v2/primitive"
"github.com/apache/rocketmq-client-go/v2/producer"
)
func main() {
// 这里按照业务逻辑来选择每个消息投递到哪个队列当中
queueSelector := producer.NewHashQueueSelector()
p, _ := rocketmq.NewProducer(
// 配置RocketMQ实例的接入点。
producer.WithNsResolver(primitive.NewPassthroughResolver([]string{"http://rocketmq-cnoea09856be****.rocketmq.volces.com:9876"})),
producer.WithRetry(2),
// 配置RocketMQ实例的AccessKey ID和AccessKey Secret。
producer.WithCredentials(primitive.Credentials{
AccessKey: "ACCCESSKEY",
SecretKey: "SECRETKEY",
}),
// 配置使用的队列选择回调
producer.WithQueueSelector(queueSelector),
)
err := p.Start()
if err != nil {
fmt.Printf("start producer error: %s", err.Error())
os.Exit(1)
}
// 配置Topic名称。对于实例 ID 格式为 MQ_INST_xxxx 的实例，此处配置的格式为${实例ID%TopicID}，例如 "MQ_INST_****%testTopic"。详细说明请参考https://www.volcengine.com/docs/6410/153010#注意事项。
topic := "testTopic"
for i := 0; i < 10; i++ {
msg := &primitive.Message{
Topic: topic,
Body: []byte("Hello World " + strconv.Itoa(i)),
}
// msg 指定分区key, 自定义即可
msg.WithShardingKey("key" + strconv.Itoa(i))
res, err := p.SendSync(context.Background(), msg)
if err != nil {
fmt.Printf("send message error: %s\n", err)
} else {
fmt.Printf("send message success: result=%s\n", res.String())
}
}
err = p.Shutdown()
if err != nil {
fmt.Printf("shutdown producer error: %s", err.Error())
}
}
订阅顺序消息
订阅顺序消息的示例代码如下。
说明
订阅顺序消息之前，需要在创建 RocketMQ Consumer 的时候配置 ConsumerOrder 为 true。
import (
"context"
"fmt"
"os"
"time"
"github.com/apache/rocketmq-client-go/v2"
"github.com/apache/rocketmq-client-go/v2/consumer"
"github.com/apache/rocketmq-client-go/v2/primitive"
)
func main() {
c, _ := rocketmq.NewPushConsumer(
//配置使用的Group ID。对于实例 ID 格式为 MQ_INST_xxxx 的实例，此处配置的格式为${实例ID%GroupID}，例如 "MQ_INST_****%demo-group"。详细说明请参考https://www.volcengine.com/docs/6410/153010#注意事项。
consumer.WithGroupName("demo-group"),
consumer.WithNsResolver(primitive.NewPassthroughResolver([]string{
"http://rocketmq-cnoea09856be****.rocketmq.volces.com:9876",
})),
// 配置RocketMQ实例的AccessKey ID和AccessKey Secret。
consumer.WithCredentials(primitive.Credentials{
AccessKey: "ACCESSKEY",
SecretKey: "SECRETKEY",
}),
consumer.WithConsumerModel(consumer.Clustering),
// 使用顺序消费的模式
consumer.WithConsumerOrder(true),
)
// 配置Topic名称。对于实例 ID 格式为 MQ_INST_xxxx 的实例，此处配置的格式为${实例ID%TopicID}，例如 "MQ_INST_****%testTopic"。详细说明请参考https://www.volcengine.com/docs/6410/153010#注意事项。
topic := "testTopic"
err := c.Subscribe(topic, consumer.MessageSelector{}, func(ctx context.Context,
msgs ...*primitive.MessageExt) (consumer.ConsumeResult, error) {
orderlyCtx, _ := primitive.GetOrderlyCtx(ctx)
fmt.Printf("orderlyCtx: %#v\n", orderlyCtx)
fmt.Printf("msgs: %v \n", msgs)
return consumer.ConsumeSuccess, nil
// 如果消费失败或者消费异常
// return consumer.SuspendCurrentQueueAMoment, err
})
if err != nil {
fmt.Println(err.Error())
}
// Note: start after subscribe
err = c.Start()
if err != nil {
fmt.Println(err.Error())
os.Exit(-1)
}
time.Sleep(time.Hour)
err = c.Shutdown()
if err != nil {
fmt.Printf("Shutdown Consumer error: %s", err.Error())
}
}
最近更新时间：2025.07.09 16:22:41
这个页面对您有帮助吗？
有用
有用
无用
无用