go语言使用RocketMq的示例代码

brokerClusterName = DefaultCluster
brokerName = broker-a
brokerId = 0
deleteWhen = 04
fileReservedTime = 48
brokerRole = ASYNC_MASTER
flushDiskType = ASYNC_FLUSH

import (
	"context"
	"fmt"
	"os"

	"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() {
	// 创建RocketMQ生产者实例
	// 参数说明:
	// - producer.WithNsResolver: 设置NameServer地址解析器，指定NameServer地址为"127.0.0.1:9876"
	// - producer.WithRetry: 设置消息发送失败时的重试次数为2次
	p, err := rocketmq.NewProducer(
		producer.WithNsResolver(primitive.NewPassthroughResolver([]string{"127.0.0.1:9876"})),
		producer.WithRetry(2),
	)
	if err != nil {
		fmt.Printf("create producer error: %s\n", err.Error())
		os.Exit(1)
	}

	err = p.Start()
	if err != nil {
		fmt.Printf("start producer error: %s\n", err.Error())
		os.Exit(1)
	}
	// 延迟执行生产者关闭操作，确保在main函数结束前关闭生产者连接并释放资源
	defer p.Shutdown()

	// 发送同步消息到RocketMQ服务器
	// 构建消息对象，指定主题和消息内容
	msg := &primitive.Message{
		Topic: "TestTopic",
		Body:  []byte("Hello RocketMQ from Go!"),
	}

	// 使用生产者同步发送消息
	// 参数说明:
	// - context.Background(): 使用空上下文进行消息发送
	// - msg: 要发送的消息对象
	// 返回值说明:
	// - res: 消息发送结果，包含消息ID等信息
	// - err: 发送过程中可能发生的错误
	res, err := p.SendSync(context.Background(), msg)//异步发送消息使用的是 SendAsync 方法
	if err != nil {
		fmt.Printf("send message error: %s\n", err)
	} else {
		fmt.Printf("send message success: %s\n", res.String())
	}

}

import (
	"context"
	"fmt"
	"os"

	"github.com/apache/rocketmq-client-go/v2"
	"github.com/apache/rocketmq-client-go/v2/consumer"
	"github.com/apache/rocketmq-client-go/v2/primitive"
)

func main() {
	// 创建RocketMQ推模式消费者实例
	// 参数说明:
	// - consumer.WithGroupName: 设置消费者组名称为"TestGroup"
	// - consumer.WithNsResolver: 设置NameServer地址解析器，指定NameServer地址为"127.0.0.1:9876"
	// 返回值说明:
	// - c: 创建的PushConsumer实例
	// - err: 创建过程中可能发生的错误
	c, err := rocketmq.NewPushConsumer(
		consumer.WithGroupName("TestGroup"),
		consumer.WithNsResolver(primitive.NewPassthroughResolver([]string{"127.0.0.1:9876"})),
	)
	if err != nil {
		fmt.Printf("create consumer error: %s\n", err.Error())
		os.Exit(1)
	}

	// 订阅指定主题的消息
	// 参数说明:
	// - "TestTopic": 要订阅的主题名称
	// - consumer.MessageSelector{}: 消息选择器，此处使用默认选择器
	// - func(ctx context.Context, msgs ...*primitive.MessageExt) (consumer.ConsumeResult, error): 消息处理回调函数
	//   - ctx: 上下文参数
	//   - msgs: 接收到的消息列表
	//   - 返回值: 消费结果和可能的错误
	// 返回值说明:
	// - err: 订阅过程中可能发生的错误
	err = c.Subscribe("TestTopic", consumer.MessageSelector{}, func(ctx context.Context,
		msgs ...*primitive.MessageExt) (consumer.ConsumeResult, error) {
		for _, msg := range msgs {
			fmt.Printf("Received message: %s\n", string(msg.Body))
		}
		return consumer.ConsumeSuccess, nil
	})
	if err != nil {
		fmt.Printf("subscribe error: %s\n", err.Error())
		os.Exit(1)
	}

	err = c.Start()
	if err != nil {
		fmt.Printf("start consumer error: %s\n", err.Error())
		os.Exit(1)
	}
	defer c.Shutdown()

	// 阻塞主 goroutine，防止程序退出
	select {}
}