在Golang中使用RocketMQ的SDK实现对特定topic的消息回调，可以按照以下步骤进行操作：
首先，确保已安装并配置好RocketMQ Go SDK。你可以通过执行
go get github.com/apache/rocketmq-client-go/v2来获取最新版本。在代码中导入相关包：
import (
"context"
"fmt"
"github.com/apache/rocketmq-client-go/v2"
"github.com/apache/rocketmq-client-go/v2/consumer"
)
- 定义回调函数，用于处理不同topic的消息逻辑：
func callbackA(ctx context.Context, msgs ...*primitive.MessageExt) (consumer.ConsumeResult, error) {
// 处理 topic A 的逻辑
for _, msg := range msgs {
fmt.Println("Received message from topic A:", string(msg.Body))
}
return consumer.ConsumeSuccess, nil
}
func callbackB(ctx context.Context, msgs ...*primitive.MessageExt) (consumer.ConsumeResult, error) {
// 处理 topic B 的逻辑
for _, msg := range msgs {
fmt.Println("Received message from topic B:", string(msg.Body))
}
return consumer.ConsumeSuccess, nil
}
- 创建Consumer，并为不同的topic注册对应的回调函数：
func main() {
namesrv := "localhost:9876" // RocketMQ服务器地址和端口号
c, _ := rocketmq.NewPushConsumer(
consumer.WithNameServer([]string{namesrv}),
)
errA := c.Subscribe("TopicA", consumer.MessageSelector{}, callbackA)
if errA != nil {
fmt.Println("Failed to subscribe to TopicA:", errA)
return
}
errB := c.Subscribe("TopicB", consumer.MessageSelector{}, callbackB)
if errB != nil {
fmt.Println("Failed to subscribe to TopicB:", errB)
return
}
err := c.Start()
if err != nil {
fmt.Printf("Consumer error: %s\n", err.Error())
return
}
select {}
}
在上面的代码中，我们使用NewPushConsumer创建一个消费者实例，并通过Subscribe方法为不同的topic注册对应的回调函数。其中，”TopicA”和”TopicB”是示例中的两个特定topic。你可以根据需要添加更多订阅。
最后，使用Start()启动消费者并进入主循环（这里使用了空的select语句，保持程序运行）。当有消息到达时，将触发相应的回调函数进行处理。
以上就是使用RocketMQ Go SDK实现对于特定topic消息回调的基本步骤。根据实际情况和需求，你可以扩展和调整代码来满足你的具体业务逻辑。
内容由零声教学AI助手提供，问题来源于学员提问
res.jsonValue["History"]..." />
res.jsonValue["History"];
..." />
