Help Center/ Distributed Message Service for RocketMQ/ Developer Guide/ Go/ Sending and Receiving Ordered Messages
Updated on 2024-03-06 GMT+08:00
View PDF

Sending and Receiving Ordered Messages

In DMS for RocketMQ, ordered messages are retrieved in the exact order that they are created.

Ordered messages are ordered globally or on the partition level.

  • Globally ordered messages: There is only one queue in a specific topic. All messages in the queue will be published and subscribed to in the first in, first out (FIFO) order.
  • Partition-level ordered message: Messages within a queue in a specific topic are published and subscribed to in the FIFO order. The producer specifies a partition selection algorithm to ensure that the messages to be ordered are allocated to the same queue.

The only difference between globally ordered messages and partition-level ordered messages is the number of queues. The code is the same.

Before sending and receiving ordered messages, collect RocketMQ connection information by referring to Collecting Connection Information.

Preparing the Environment

  1. Run the following command to check whether Go has been installed: 
    go version

    If the following information is displayed, Go has been installed:

    go version go1.16.5 linux/amd64

    If Go is not installed, download and install it.

  2. Add the following code to go.mod to add the dependency: 
    module rocketmq-example-go

go 1.13

require (
	github.com/apache/rocketmq-client-go/v2 v2.1.1
)

Sending Ordered Messages

The following code is an example. Replace the information in bold with the actual values.

package main

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"
)

// Package main implements a simple producer to send message.
func main() {
	p, _ := rocketmq.NewProducer(
		producer.WithNsResolver(primitive.NewPassthroughResolver([]string{"192.168.0.1:8100"})),
		producer.WithRetry(2),
	)
	err := p.Start()
	if err != nil {
		fmt.Printf("start producer error: %s", err.Error())
		os.Exit(1)
	}
	topic := "test"

	for i := 0; i < 100; i++ {
		msg := &primitive.Message{
			Topic: topic,
			Body:  []byte("Hello RocketMQ Go Client! " + strconv.Itoa(i)),
		}
		orderId := strconv.Itoa(i % 10)
		msg.WithShardingKey(orderId)
		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())
	}
}

The parameters in the example code are described as follows. For details about how to obtain the parameter values, see Collecting Connection Information.

  • 192.168.0.1:8100: instance metadata address and port.
  • test: topic name.

In the preceding code, to ensure the sequence of messages with the same orderId, orderId is used as the sharding key of the specific queue.

Subscribing to Ordered Messages

You only need to add consumer.WithConsumerOrder(true) to the code for subscribing to normal messages. The following code is an example. Replace the information in bold with the actual values.

package main

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(
		consumer.WithGroupName("testGroup"),
		consumer.WithNsResolver(primitive.NewPassthroughResolver([]string{"192.168.0.1:8100"})),
		consumer.WithConsumerModel(consumer.Clustering),
		consumer.WithConsumeFromWhere(consumer.ConsumeFromFirstOffset),
		consumer.WithConsumerOrder(true),
	)
	err := c.Subscribe("test", consumer.MessageSelector{}, func(ctx context.Context,
		msgs ...*primitive.MessageExt) (consumer.ConsumeResult, error) {
		orderlyCtx, _ := primitive.GetOrderlyCtx(ctx)
		fmt.Printf("orderly context: %v\n", orderlyCtx)
		fmt.Printf("subscribe orderly callback: %v \n", msgs)
		return consumer.ConsumeSuccess, nil
	})
	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())
	}
}

The parameters in the example code are described as follows. For details about how to obtain the parameter values, see Collecting Connection Information.

  • testGroup: consumer group name.
  • 192.168.0.1:8100: instance metadata address and port.
  • test: topic name.

Parent topic: Go