英文网站推广工作,深圳制作网站有几家,网页制作 视频教程,上海公司名义买房条件上篇文章《RocketMQ消息队列#xff08;一#xff09;—— 基本概念和消息类型》记录了RocketMQ的一些基本的概念#xff0c;本文主要写几个go语言操作RocketMQ的示例代码
一、发送普通消息
import (contextfmtgithub.com/apache/rocketmq-… 上篇文章《RocketMQ消息队列一—— 基本概念和消息类型》记录了RocketMQ的一些基本的概念本文主要写几个go语言操作RocketMQ的示例代码
一、发送普通消息
import (contextfmtgithub.com/apache/rocketmq-client-go/v2github.com/apache/rocketmq-client-go/v2/primitivegithub.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 (contextfmtgithub.com/apache/rocketmq-client-go/v2github.com/apache/rocketmq-client-go/v2/consumergithub.com/apache/rocketmq-client-go/v2/primitivetime
)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 (contextfmtgithub.com/apache/rocketmq-client-go/v2github.com/apache/rocketmq-client-go/v2/primitivegithub.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 (contextfmtostimegithub.com/apache/rocketmq-client-go/v2github.com/apache/rocketmq-client-go/v2/primitivegithub.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())}
} 后记 个人总结欢迎转载、评论、批评指正
