操作场景
本文以调用 Go SDK 为例介绍通过开源 SDK 实现消息收发的操作过程,帮助您更好地理解消息收发的完整过程。
说明:
以 Go 客户端为例说明,其他语言客户端请参见 SDK 文档。
前提条件
完成资源创建与准备
下载 Demo
操作步骤
步骤1:安装 Go 依赖库
在 Golang 项目中引入相关依赖,以
go get
为例,使用如下命令:go get github.com/apache/rocketmq-clients/golang/v5
步骤2:生产消息
package mainimport ("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 添加配置的akAccessKey = "xxxxxx"// SecretKey 添加配置的skSecretKey = "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 producererr = producer.Start()if err != nil {log.Fatal(err)}// graceful stop producerdefer producer.GracefulStop()for i := 0; i < 10; i++ {// new a messagemsg := &rmq_client.Message{Topic: Topic,Body: []byte("this is a message : " + strconv.Itoa(i)),}// set keys and tagmsg.SetKeys("a", "b")msg.SetTag("ab")// send message in syncresp, 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 momenttime.Sleep(time.Second * 1)}}
步骤3:消费消息
腾讯云消息队列 TDMQ RocketMQ 版 5.x 系列支持两种消费模式,分别为 Push Consumer 和 Simple Consumer。
说明:
社区版 Golang SDK 目前仅支持 Simple Consumer。
以下代码示例以 Simple Consumer 为例:
package mainimport ("context""fmt""log""os""os/signal" "sync" "syscall""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 添加配置的akAccessKey = "xxxxxx"// SecretKey 添加配置的skSecretKey = "xxxxxx")var (// maximum waiting time for receive funcawaitDuration = time.Second * 5// maximum number of messages received at one timemaxMessageNum int32 = 16// invisibleDuration should > 20sinvisibleDuration = time.Second * 20// receive concurrency receiveConcurrency = 6)func main() {// log to consoleos.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 simpleConsumererr = simpleConsumer.Start()if err != nil {log.Fatal(err)}// graceful stop simpleConsumerdefer 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:returndefault:mvs, err := simpleConsumer.Receive(context.TODO(), maxMessageNum, invisibleDuration)if err != nil {fmt.Println("receive message error: " + err.Error())}// ack messagefor _, 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<-exitclose(ch)wg.Wait()}
步骤4:查看消息详情