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() {
	// 創(chuàng)建RocketMQ生產(chǎn)者實(shí)例
	// 參數(shù)說(shuō)明:
	// - producer.WithNsResolver: 設(shè)置NameServer地址解析器，指定NameServer地址為"127.0.0.1:9876"
	// - producer.WithRetry: 設(shè)置消息發(fā)送失敗時(shí)的重試次數(shù)為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)
	}
	// 延遲執(zhí)行生產(chǎn)者關(guān)閉操作，確保在main函數(shù)結(jié)束前關(guān)閉生產(chǎn)者連接并釋放資源
	defer p.Shutdown()

	// 發(fā)送同步消息到RocketMQ服務(wù)器
	// 構(gòu)建消息對(duì)象，指定主題和消息內(nèi)容
	msg := &primitive.Message{
		Topic: "TestTopic",
		Body:  []byte("Hello RocketMQ from Go!"),
	}

	// 使用生產(chǎn)者同步發(fā)送消息
	// 參數(shù)說(shuō)明:
	// - context.Background(): 使用空上下文進(jìn)行消息發(fā)送
	// - msg: 要發(fā)送的消息對(duì)象
	// 返回值說(shuō)明:
	// - res: 消息發(fā)送結(jié)果，包含消息ID等信息
	// - err: 發(fā)送過(guò)程中可能發(fā)生的錯(cuò)誤
	res, err := p.SendSync(context.Background(), msg)//異步發(fā)送消息使用的是 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() {
	// 創(chuàng)建RocketMQ推模式消費(fèi)者實(shí)例
	// 參數(shù)說(shuō)明:
	// - consumer.WithGroupName: 設(shè)置消費(fèi)者組名稱為"TestGroup"
	// - consumer.WithNsResolver: 設(shè)置NameServer地址解析器，指定NameServer地址為"127.0.0.1:9876"
	// 返回值說(shuō)明:
	// - c: 創(chuàng)建的PushConsumer實(shí)例
	// - err: 創(chuàng)建過(guò)程中可能發(fā)生的錯(cuò)誤
	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)
	}

	// 訂閱指定主題的消息
	// 參數(shù)說(shuō)明:
	// - "TestTopic": 要訂閱的主題名稱
	// - consumer.MessageSelector{}: 消息選擇器，此處使用默認(rèn)選擇器
	// - func(ctx context.Context, msgs ...*primitive.MessageExt) (consumer.ConsumeResult, error): 消息處理回調(diào)函數(shù)
	//   - ctx: 上下文參數(shù)
	//   - msgs: 接收到的消息列表
	//   - 返回值: 消費(fèi)結(jié)果和可能的錯(cuò)誤
	// 返回值說(shuō)明:
	// - err: 訂閱過(guò)程中可能發(fā)生的錯(cuò)誤
	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 {}
}