How to use

go mod

require ( v2.0.0-alpha1

Set Logger

Go Client define the Logger interface for log output, user can specify implementation of private. in default, client use logrus.


Send message


Producer interface {
	Start() error
	Shutdown() error
	SendSync(context.Context, *primitive.Message) (*internal.SendResult, error)
	SendOneWay(context.Context, *primitive.Message) error


  • create a new Producer instance
opt := producer.ProducerOptions{
    NameServerAddr:           "",
    RetryTimesWhenSendFailed: 2,
p := producer.NewProducer(opt)
  • start the producer
err := p.Start()
  • send message with sync
result, err := p.SendSync(context.Background(), &primitive.Message{
    Topic: "test",
    Body:  []byte("Hello RocketMQ Go Client!"),

// do something with result
  • or send message with oneway
err := p.SendOneWay(context.Background(), &primitive.Message{
    Topic: "test",
    Body:  []byte("Hello RocketMQ Go Client!"),

Full examples: producer

Consume Message

alpha1 only support PushConsumer


PushConsumer interface {
	Start() error
	Subscribe(topic string, selector MessageSelector,
		f func(*ConsumeMessageContext, []*primitive.MessageExt) (ConsumeResult, error)) error


  • Create a PushConsumer instance
c := consumer.NewPushConsumer("testGroup", consumer.ConsumerOption{
    NameServerAddr: "",
    ConsumerModel:  consumer.Clustering,
    FromWhere:      consumer.ConsumeFromFirstOffset,
  • Subscribe a topic(only support one topic now), and define your consuming function
err := c.Subscribe("test", consumer.MessageSelector{}, func(ctx *consumer.ConsumeMessageContext,
    msgs []*primitive.MessageExt) (consumer.ConsumeResult, error) {
    return consumer.ConsumeSuccess, nil
  • start the consumer(NOTE: MUST after subscribe)

Full examples: consumer