golang操作rocketmq的示例代码
短信预约 -IT技能 免费直播动态提醒
RocketMQ 是什么
Github 上关于 RocketMQ 的介绍：
RcoketMQ 是一款低延迟、高可靠、可伸缩、易于使用的消息中间件。具有以下特性：
- 支持发布/订阅（Pub/Sub）和点对点（P2P）消息模型
- 在一个队列中可靠的先进先出（FIFO）和严格的顺序传递
- 支持拉（pull）和推（push）两种消息模式
- 单一队列百万消息的堆积能力
- 支持多种消息协议，如 JMS、MQTT 等
- 分布式高可用的部署架构,满足至少一次消息传递语义
- 提供 docker 镜像用于隔离测试和云集群部署
- 提供配置、指标和监控等功能丰富的 Dashboard
对于这些特性描述，大家简单过一眼就即可，深入学习之后自然就明白了。
下面看下golang操作rocketmq的示例代码，内容如下所示：
下载
go get github.com/apache/rocketmq-client-go/v2代码
func main() {
// 1. 创建主题
//CreateTopic("test-04", 10909)
// 2. 生产者向主题中发送消息
//SendSyncMessage("hello world0002")
// 3. 消费者订阅主题并消费
SubscribeMessage()
}
func CreateTopic(topicName string, port int) {
// 创建主题
testAdmin, err := admin.NewAdmin(admin.WithResolver(primitive.NewPassthroughResolver([]string{"ip:server_port"})))
if err != nil {
fmt.Println(err)
}
err = testAdmin.CreateTopic(
context.Background(),
admin.WithTopicCreate(topicName),
admin.WithBrokerAddrCreate(fmt.Sprintf("ip:%d", port)),
)
fmt.Println(err)
func SendSyncMessage(message string) {
endPoint := []string{"ip:server_port"}
p, err := rocketmq.NewProducer(
producer.WithNameServer(endPoint),
//producer.WithNsResolver(primitive.NewPassthroughResolver(endPoint)),
producer.WithRetry(2),
err = p.Start()
result, err := p.SendSync(context.Background(), &primitive.Message{
Topic: "test",
Body: []byte(message),
})
fmt.Println(result.Status, result)
func SubscribeMessage() {
// 订阅主题、并消费
c, err := rocketmq.NewPushConsumer(
consumer.WithNameServer(endPoint),
consumer.WithConsumerModel(consumer.Clustering),
consumer.WithGroupName("GID_TEST01"),
//fmt.Println(err)
err = c.Subscribe("test", consumer.MessageSelector{}, func(ctx context.Context,
msgs ...*primitive.MessageExt) (consumer.ConsumeResult, error) {
for i := range msgs {
fmt.Printf("subscribe callback: %v \n", msgs[i])
}
return consumer.ConsumeSuccess, nil
//fmt.Println(err.Error())
// Note: start after subscribe
err = c.Start()
os.Exit(-1)
c.Shutdown()到此这篇关于golang操作rocketmq的示例代码的文章就介绍到这了,更多相关golang操作rocketmq内容请搜索编程网以前的文章或继续浏览下面的相关文章希望大家以后多多支持编程网！
免责声明：
① 本站未注明“稿件来源”的信息均来自网络整理。其文字、图片和音视频稿件的所属权归原作者所有。本站收集整理出于非商业性的教育和科研之目的，并不意味着本站赞同其观点或证实其内容的真实性。仅作为临时的测试数据，供内部测试之用。本站并未授权任何人以任何方式主动获取本站任何信息。
② 本站未注明“稿件来源”的临时测试数据将在测试完成后最终做删除处理。有问题或投稿请发送至: 邮箱/279061341@qq.com QQ/279061341
