Sending and Receiving Ordered Messages

Notes and Constraints

To receive and send orderly messages, ensure the topic message type is Orderly before connecting a client to a RocketMQ instance of v5.x.

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. Go to the bin directory where the Go script is in.
3. Run the touch go.mod command to create a go.mod file and add the following code to it to add the dependency:

   module rocketmq-example-go
   
   go 1.13
   
   require (
   	github.com/apache/rocketmq-client-go/v2 v2.1.2
   )

4. Run the following command to download the dependency:

   go mod tidy

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())
	}
}

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 address and port.
• test: topic name.