分布式消息服务ROCKETMQ版-开启和查询RocketMQ消息轨迹:开启RocketMQ消息轨迹（Go）

时间：2025-03-07 09:45:18

开启RocketMQ消息轨迹（Go）

在客户端开启消息轨迹的方法如下：

执行以下命令，检查是否已安装Go。
```
go version
```
返回如下回显时，说明Go已经安装。
```
[root@ecs-test sarama]# go version
go version go1.16.5 linux/amd64
```
如果未安装Go，请下载并安装。
进入Go脚本所在的bin目录下。

执行“touch go.mod”命令新建一个“go.mod”，并增加以下代码，添加依赖。

module rocketmq-example-go

go 1.13

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

生产者开启消息轨迹（以下加粗内容需要替换为实例自有信息，请根据实际情况替换）。

package main

import (
	"context"
	"fmt"
	"os"
	"time"

	"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() {
	namesrvs := []string{"192.168.0.1:8100"}
	traceCfg := &primitive.TraceConfig{
		Access:   primitive.Local,
		Resolver: primitive.NewPassthroughResolver(namesrvs),
	}

	p, _ := rocketmq.NewProducer(
		producer.WithNsResolver(primitive.NewPassthroughResolver([]string{"192.168.0.1:8100"})),
		producer.WithRetry(2),
		producer.WithTrace(traceCfg))   // 增加此行代码表示开启了消息轨迹。
	err := p.Start()
	if err != nil {
		fmt.Printf("start producer error: %s", err.Error())
		os.Exit(1)
	}
	res, err := p.SendSync(context.Background(), primitive.NewMessage("topic1",
		[]byte("Hello RocketMQ Go Client!")))

	if err != nil {
		fmt.Printf("send message error: %s\n", err)
	} else {
		fmt.Printf("send message success: result=%s\n", res.String())
	}

	time.Sleep(10 * time.Second)

	err = p.Shutdown()
	if err != nil {
		fmt.Printf("shutdown producer error: %s", err.Error())
	}
}

消费者开启消息轨迹（以下加粗内容需要替换为实例自有信息，请根据实际情况替换）。

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() {
	namesrvs := []string{"192.168.0.1:8100"}
	traceCfg := &primitive.TraceConfig{
		Access:   primitive.Local,
		Resolver: primitive.NewPassthroughResolver(namesrvs),
	}

	c, _ := rocketmq.NewPushConsumer(
		consumer.WithGroupName("testGroup"),
		consumer.WithNsResolver(primitive.NewPassthroughResolver([]string{"192.168.0.1:8100"})),
		consumer.WithTrace(traceCfg),   // 增加此行代码表示开启了消息轨迹。
	)
	err := c.Subscribe("TopicTest", consumer.MessageSelector{}, func(ctx context.Context,
		msgs ...*primitive.MessageExt) (consumer.ConsumeResult, error) {
		fmt.Printf("subscribe 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())
	}
}

上一篇：分布式消息服务ROCKETMQ版-开启和查询RocketMQ消息轨迹:开启RocketMQ消息轨迹（Java）

下一篇：分布式消息服务ROCKETMQ版-开启和查询RocketMQ消息轨迹:开启RocketMQ消息轨迹（Java）