Message Queue for RocketMQ
Message Queue for RocketMQ
Document
- Documentation
Message Queue for RocketMQSDK ReferenceRocketMQ 5.xGO 5.x SDKOrdered message
Copy page
Download PDFGO 5.x SDK
Ordered message
Copy pageDownload PDFOrdered message
Message Queue RocketMQ 5.x provides ordered messages (FIFO messages). In the ordered message model, you must strictly publish and consume messages in sequence. This topic provides sample code for sending and receiving ordered messages using the Go SDK.
Background information
- Before using ordered messages, create a topic and group of ordered message type in the console.
- Ordered messages are divided into two types: global ordered messages and partitioned ordered messages. The only difference is the number of queues; the code remains the same.
- Global order:
- For a specified topic, all message production and consumption must follow a strict order. The consumption order must match the production order, requiring strict First In First Out (FIFO) publishing and consumption.
- Partition order:
- For a specified Topic, message production and consumption within each partition are ordered. Messages in the same queue are published and subscribed strictly in FIFO order. The messageGroup distinguishes which partition a message is delivered to.
Send ordered messages
The following sample code sends ordered messages.
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"
)
// Set the endpoint information you obtained from the VolcanoEngine Message Queue RocketMQ console, such as "{INSTANCE_ID}.rocketmq.ivolces.com:8080".
// Enter the corresponding authentication information based on the authentication methods supported by the instance:
// Key management: AccessKey refers to AccessKey ID, and SecretKey refers to AccessKey Secret.
// Permission management: AccessKey refers to the ACL username, and SecretKey refers to the ACL password.
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)
}
}
Subscribe to ordered messages
The following sample code subscribes to ordered messages. Before use, confirm that the Group type supports ordered messages.
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"
)
// Set the endpoint from the VolcanoEngine Message Queue RocketMQ console, such as "{INSTANCE_ID}.rocketmq.ivolces.com:8080".
// Enter the authentication information according to the authentication methods supported by the instance.
// Key management: AccessKey refers to the AccessKey ID, and SecretKey refers to the AccessKey Secret.
// Permission management: AccessKey refers to the ACL username, and SecretKey refers to the ACL password.
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)
}
Last updated: 2026.08.26 15:56:31
All rights reserved © Beijing Volcano Engine Technology Co., Ltd2024
