消息队列 RocketMQ版
消息队列 RocketMQ版
文档指南
请输入
- 文档首页
消息队列 RocketMQ版SDK 参考RocketMQ 5.xGo 5.x SDK顺序消息
复制全文
下载 pdf顺序消息
文档反馈
问问助手消息队列 RocketMQ 5.x 版本提供顺序消息（FIFO 消息）供您使用。在顺序消息模型中，您需要严格按照顺序来发布和消费消息。本文提供使用 Go SDK 收发顺序消息的示例代码供您参考。
背景信息
- 使用顺序消息前，需在控制台创建顺序消息类型的 Topic 和 Group。
- 顺序消息分为两类，全局顺序消息和分区顺序消息。区别仅为队列数量不同，代码没有区别。
- 全局顺序：
- 对于指定的一个 Topic，所有消息的生产和消费需要遵循一定的顺序，消息的消费顺序必须和生产顺序一致，即需要严格的先入先出 FIFO（First In First Out）的顺序进行发布和消费。
- 分区顺序：
- 对于指定的一个 Topic，其中每一个分区的消息生产与消费是有序的，同一个队列内的消息按照严格的 FIFO 顺序进行发布和订阅。消息投递到哪一个分区由消息的messageGroup来进行区分。
发送顺序消息
发送顺序消息的示例代码如下。
package main
import (
"context"
"fmt"
"log"
"os"
"strconv"
"time"
rmq_client "github.com/apache/rocketmq-clients/golang/v5"
"github.com/apache/rocketmq-clients/golang/v5/credentials"
)
// 设置为您从火山引擎消息队列 RocketMQ版控制台获取的接入点信息，类似"{INSTANCE_ID}.rocketmq.ivolces.com:8080"。
// 根据实例支持的认证方式，填写对应的身份认证信息：
// 密钥管理：AccessKey 为 AccessKey ID，SecretKey 为 AccessKey Secret。
// 权限管理：AccessKey 为 ACL 用户名，SecretKey 为 ACL 密码。
const (
Topic = "xxxx"
Endpoint = "rocketmq-xxxx:8080"
AccessKey = "xxxx"
SecretKey = "xxxx"
)
func main() {
// log to console
os.Setenv("mq.consoleAppender.enabled", "true")
rmq_client.ResetLogger()
// In most case, you don't need to create many producers, singleton pattern is more recommended.
producer, err := rmq_client.NewProducer(&rmq_client.Config{
Endpoint: Endpoint,
Credentials: &credentials.SessionCredentials{
AccessKey: AccessKey,
AccessSecret: SecretKey,
},
},
rmq_client.WithTopics(Topic),
)
if err != nil {
log.Fatal(err)
}
// start producer
err = producer.Start()
if err != nil {
log.Fatal(err)
}
// graceful stop producer
defer producer.GracefulStop()
for i := 0; i < 10; i++ {
// new a message
msg := &rmq_client.Message{
Topic: Topic,
Body: []byte("this is a message : " + strconv.Itoa(i)),
}
// set keys and tag
msg.SetKeys("a", "b")
msg.SetTag("ab")
// Message group decides the message delivery order.
msg.SetMessageGroup("fifo")
// send message in sync
resp, err := producer.Send(context.TODO(), msg)
if err != nil {
log.Fatal(err)
}
for i := 0; i < len(resp); i++ {
fmt.Printf("%#v\n", resp[i])
}
// wait a moment
time.Sleep(time.Second * 1)
}
}
订阅顺序消息
订阅顺序消息的示例代码如下。使用前请确认所用 Group 为顺序消息类型。
package main
import (
"context"
"fmt"
"log"
"os"
"time"
rmq_client "github.com/apache/rocketmq-clients/golang/v5"
"github.com/apache/rocketmq-clients/golang/v5/credentials"
)
// 设置为您从火山引擎消息队列 RocketMQ版控制台获取的接入点信息，类似"{INSTANCE_ID}.rocketmq.ivolces.com:8080"。
// 根据实例支持的认证方式，填写对应的身份认证信息：
// 密钥管理：AccessKey 为 AccessKey ID，SecretKey 为 AccessKey Secret。
// 权限管理：AccessKey 为 ACL 用户名，SecretKey 为 ACL 密码。
const (
Topic = "xxxx"
ConsumerGroup = "xxxx"
Endpoint = "rocketmq-xxxx:8080"
AccessKey = "xxxx"
SecretKey = "xxxx"
)
var (
// maximum waiting time for receive func
awaitDuration = time.Second * 5
// maximum number of messages received at one time
maxMessageNum int32 = 16
// invisibleDuration should > 20s
invisibleDuration = time.Second * 20
// receive messages in a loop
receiveConcurrency = 4
)
func main() {
// log to console
os.Setenv("mq.consoleAppender.enabled", "true")
rmq_client.ResetLogger()
// In most case, you don't need to create many consumers, singleton pattern is more recommended.
simpleConsumer, err := rmq_client.NewSimpleConsumer(&rmq_client.Config{
Endpoint: Endpoint,
ConsumerGroup: ConsumerGroup,
Credentials: &credentials.SessionCredentials{
AccessKey: AccessKey,
AccessSecret: SecretKey,
},
},
rmq_client.WithAwaitDuration(awaitDuration),
rmq_client.WithSubscriptionExpressions(map[string]*rmq_client.FilterExpression{
Topic: rmq_client.SUB_ALL,
}),
)
if err != nil {
log.Fatal(err)
}
// start simpleConsumer
err = simpleConsumer.Start()
if err != nil {
log.Fatal(err)
}
// graceful stop simpleConsumer
defer simpleConsumer.GracefulStop()
go func() {
for {
fmt.Println("start receive message")
mvs, err := simpleConsumer.Receive(context.TODO(), maxMessageNum, invisibleDuration)
if err != nil {
fmt.Println(err)
}
// ack message
for _, mv := range mvs {
simpleConsumer.Ack(context.TODO(), mv)
fmt.Println(mv)
}
fmt.Println("wait a moment")
fmt.Println()
}
}()
// run for a while
time.Sleep(100 * time.Minute)
}
最近更新时间：2026.08.26 15:19:55
这个页面对您有帮助吗？
有用
有用
无用
无用