RocketMQ消息队列（二）—— Go语言操作RocketMQ

上篇文章《RocketMQ消息队列（一）—— 基本概念和消息类型》记录了RocketMQ的一些基本的概念，本文主要写几个go语言操作RocketMQ的示例代码

一、发送普通消息

import ("context""fmt""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() {p, err := rocketmq.NewProducer(producer.WithNsResolver(primitive.NewPassthroughResolver([]string{"127.0.0.1:9876"})),producer.WithRetry(2), //指定重试次数)if err != nil {panic(err)}if err = p.Start(); err != nil {panic("启动producer失败")}topic := "test"// 构建一个消息message := primitive.NewMessage(topic, []byte("hello world!"))res, err := p.SendSync(context.Background(), message)if err != nil {fmt.Printf("send message error: %s\n", err)} else {fmt.Printf("send message success: result=%s\n", res.String())}if err = p.Shutdown(); err != nil {panic("关闭producer失败")}
}

二、消费消息

import ("context""fmt""github.com/apache/rocketmq-client-go/v2""github.com/apache/rocketmq-client-go/v2/consumer""github.com/apache/rocketmq-client-go/v2/primitive""time"
)func main() {c, err := rocketmq.NewPushConsumer(consumer.WithGroupName("test"),consumer.WithNsResolver(primitive.NewPassthroughResolver([]string{"127.0.0.1:9876"})),)if err != nil {panic(err)}if err := c.Subscribe("test",consumer.MessageSelector{},// 收到消息后的回调函数func(ctx context.Context, msgs ...*primitive.MessageExt) (consumer.ConsumeResult, error) {for i := range msgs {fmt.Printf("获取到值： %v \n", msgs[i])}return consumer.ConsumeSuccess, nil}); err != nil {}err = c.Start()if err != nil {panic("启动consumer失败")}//不能让主goroutine退出time.Sleep(time.Hour)_ = c.Shutdown()
}

三、发送延迟消息

延迟消息和普通的发送区别就是在需要发送的消息上，用下面的代码设置发送的级别即可

message.WithDelayTimeLevel(3)

全部代码如下：


import ("context""fmt""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() {p, err := rocketmq.NewProducer(producer.WithNsResolver(primitive.NewPassthroughResolver([]string{"127.0.0.1:9876"})),producer.WithRetry(2), //指定重试次数)if err != nil {panic(err)}if err = p.Start(); err != nil {panic("启动producer失败")}topic := "test"// 构建一个消息message := primitive.NewMessage(topic, []byte("this is a delay message!"))// 给message设置延迟级别message.WithDelayTimeLevel(3)res, err := p.SendSync(context.Background(), message)if err != nil {fmt.Printf("send message error: %s\n", err)} else {fmt.Printf("send message success: result=%s\n", res.String())}if err = p.Shutdown(); err != nil {panic("关闭producer失败")}
}

四、发送事务消息

发送事务消息需要我们写一个TransactionListener接口的方法，指明事务执行成功和回调的具体操作，接口如下

type TransactionListener interface {//  When send transactional prepare(half) message succeed, this method will be invoked to execute local transaction.ExecuteLocalTransaction(*Message) LocalTransactionState// When no response to prepare(half) message. broker will send check message to check the transaction status, and this// method will be invoked to get local transaction status.CheckLocalTransaction(*MessageExt) LocalTransactionState
}

完整的代码如下：

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"
)type DemoListener struct {
}func (dl *DemoListener) ExecuteLocalTransaction(msg *primitive.Message) primitive.LocalTransactionState {fmt.Println("开始执行本地逻辑")time.Sleep(time.Second * 3)fmt.Println("执行本地逻辑失败")//本地执行逻辑无缘无故失败 代码异常 宕机return primitive.UnknowState
}func (dl *DemoListener) CheckLocalTransaction(msg *primitive.MessageExt) primitive.LocalTransactionState {fmt.Println("rocketmq的消息回查")return primitive.CommitMessageState
}func main() {p, _ := rocketmq.NewTransactionProducer(&DemoListener{},producer.WithNsResolver(primitive.NewPassthroughResolver([]string{"127.0.0.1:9876"})),producer.WithRetry(1),)err := p.Start()if err != nil {fmt.Printf("start producer error: %s\n", err.Error())os.Exit(1)}res, err := p.SendMessageInTransaction(context.Background(),primitive.NewMessage("TopicTest5", []byte("Hello RocketMQ again ")))if err != nil {fmt.Printf("send message error: %s\n", err)} else {fmt.Printf("send message success: result=%s\n", res.String())}time.Sleep(5 * time.Minute)err = p.Shutdown()if err != nil {fmt.Printf("shutdown producer error: %s", err.Error())}
}