消息队列 RocketMQ 版 Go SDK 使用-RocketMQ 5.x-文档中心-腾讯云

操作场景
本文以调用 Go SDK 为例介绍通过开源 SDK 实现消息收发的操作过程，帮助您更好地理解消息收发的完整过程。
说明：
以 Go 客户端为例说明，其他语言客户端请参见 SDK 文档。
前提条件
完成资源创建与准备
﻿安装 1.13 版本以上的 golang﻿
﻿下载 Demo﻿
操作步骤
步骤1：安装 Go 依赖库
在 Golang 项目中引入相关依赖，以 go get 为例，使用如下命令：
go get github.com/apache/rocketmq-clients/golang/v5
步骤2：生产消息
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"
)
﻿
const (
    Topic = "xxxxxx"
    // Endpoint 填写腾讯云提供的接入地址
    Endpoint = "xxxxxx"
    // AccessKey 添加配置的ak
    AccessKey = "xxxxxx"
    // SecretKey 添加配置的sk
    SecretKey = "xxxxxx"
)
﻿
func main() {
    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")
        // 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)
    }
}
步骤3：消费消息
腾讯云消息队列 TDMQ RocketMQ 版 5.x 系列支持两种消费模式，分别为 Push Consumer 和 Simple Consumer。
说明：
社区版 Golang SDK 目前仅支持 Simple Consumer。
以下代码示例以 Simple Consumer 为例：
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"
)
﻿
const (
    Topic         = "xxxxxx"
    ConsumerGroup = "xxxxxx"
    // Endpoint 填写腾讯云提供的接入地址
    Endpoint = "xxxxxx"
    // AccessKey 添加配置的ak
    AccessKey = "xxxxxx"
    // SecretKey 添加配置的sk
    SecretKey = "xxxxxx"
)
﻿
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
)
﻿
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 func() {
		if r := recover(); r != nil {
			fmt.Println(r)
		}
		_ = simpleConsumer.GracefulStop()
	}()
﻿
	fmt.Println("start receive message")
﻿
	// Each Receive call will only select one broker queue to pop messages.
	// Enable multiple consumption goroutines to reduce message end-to-end latency.
	ch := make(chan struct{})
	wg := &sync.WaitGroup{}
	for i := 0; i < receiveConcurrency; i++ {
		wg.Add(1)
		go func() {
			defer wg.Done()
			for {
				select {
				case <-ch:
					return
				default:
					mvs, err := simpleConsumer.Receive(context.TODO(), maxMessageNum, invisibleDuration)
					if err != nil {
						fmt.Println("receive message error: " + err.Error())
					}
					// ack message
					for _, mv := range mvs {
						fmt.Println(mv)
						if err := simpleConsumer.Ack(context.TODO(), mv); err != nil {
							fmt.Println("ack message error: " + err.Error())
						}
					}
				}
			}
		}()
	}
﻿
	exit := make(chan os.Signal, 1)
	signal.Notify(exit, syscall.SIGINT, syscall.SIGTERM)
﻿
	// wait for exit
	<-exit
	close(ch)
	wg.Wait()
}
步骤4：查看消息详情
发送完成消息后会得到一个消息 ID （messageID），开发者可以在消息查询页面查询刚刚发送的消息，如下图所示；并且可以查看特定消息的详情和轨迹等信息，详情请参见 消息查询。
﻿
﻿
﻿
Go SDK 使用

本页目录：

操作场景

前提条件

操作步骤

步骤1：安装 Go 依赖库

步骤2：生产消息

步骤3：消费消息

步骤4：查看消息详情
