收发普通消息(1) 异步发送 异步发送是指消息发送方发出一条消息后,不等服务端返回响应,接着发送下一条消息的通讯方式。 使用异步发送需要客户端实现异步发送回调接口(SendCallback)。即消息发送方在发送了一条消息后,不需要等待服务端响应接着发送第二条消息。发送方通过回调接口接收服务端响应,并处理响应结果。 参考如下示例代码。 package main import ( "context" "fmt" "github.com/apache/rocketmqclientgo/v2" "github.com/apache/rocketmqclientgo/v2/primitive" "github.com/apache/rocketmqclientgo/v2/producer" "os" "sync" ) func main() { // 填写分布式消息服务RocketMQ控制台Namesrv接入点 endpoint : "${ENDPOINT}" // 填写AccessKey,在分布式消息服务RocketMQ控制台用户管理菜单中创建的用户ID accessKey : "${ACCESSKEY}" // 填写SecretKey 在在分布式消息服务RocketMQ控制台用户管理菜单中创建的用户密钥 secretKey : "${SECRETKEY}" // 填写Topic,在管理控制台创建 topic : "${TOPIC}" p, : rocketmq.NewProducer( producer.WithNsResolver(primitive.NewPassthroughResolver([]string{endpoint})), producer.WithRetry(2), producer.WithCredentials(primitive.Credentials{ AccessKey: accessKey, SecretKey: secretKey, }), ) err : p.Start() if err ! nil { fmt.Printf("start producer error: %s", err.Error()) os.Exit(1) } // 使用异步方式发送消息 var wg sync.WaitGroup for i : 0; i < 4; i++ { wg.Add(1) err : p.SendAsync(context.TODO(), func(ctx context.Context, result primitive.SendResult, e error) { if e ! nil { fmt.Printf("send message error: %sn", err) } else { fmt.Printf("send message success: result%sn", result.String()) } wg.Done() }, primitive.NewMessage(topic, []byte("Hello RocketMQ"))) if err ! nil { fmt.Printf("send message error: %sn", err) } } wg.Wait() err p.Shutdown() if err ! nil { fmt.Printf("shutdown producer error: %s", err.Error()) } }
来自: