rocketmq

README

RocketMQ Client Go

A product ready RocketMQ Client in pure go, which supports almost the full features of Apache RocketMQ, such as pub and sub messages, ACL, tracing and so on.

---

Due Diligence

Here, we sincerely invite you to take a minute to feedback on your usage scenario.

---

Features

For 2.X version, it supports:

• sending message in synchronous mode
• sending message in asynchronous mode
• sending message in oneway mode
• sending message in batch mode
• sending orderly messages
• sending delay messages
• sending transaction messages
• consuming message using push model
• consuming message using pull model
• consuming message using broadcast model
• message tracing for pub and sub messages
• ACL for producers and consumers
• request-reply model

---

How to use

• Step-by-step instruction are provided in RocketMQ Go Client Introduction
• Consult RocketMQ Quick Start to setup rocketmq broker and nameserver.

---

Apache RocketMQ Community

• RocketMQ Community Projects

---

Contact us

• Mailing Lists: https://rocketmq.apache.org/about/contact/
• Home: https://rocketmq.apache.org
• Docs: https://rocketmq.apache.org/docs/quick-start/
• Issues: https://github.com/apache/rocketmq-client-go/issues
• Ask: https://stackoverflow.com/questions/tagged/rocketmq
• Slack: https://rocketmq-community.slack.com/

---

How to Contribute

Contributions are warmly welcome! Be it trivial cleanup, major new feature or other suggestion. Read this how to contribute guide for more details.

---

License

Apache License, Version 2.0 Copyright (C) Apache Software Foundation

Documentation

Overview

Licensed to the Apache Software Foundation (ASF) under one or more contributor license agreements. See the NOTICE file distributed with this work for additional information regarding copyright ownership. The ASF licenses this file to You under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with the License. You may obtain a copy of the License at

http://www.apache.org/licenses/LICENSE-2.0

Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the specific language governing permissions and limitations under the License.

Index

• type Producer
  • func NewProducer(opts ...producer.Option) (Producer, error)
• type PullConsumer
  • func NewPullConsumer(opts ...consumer.Option) (PullConsumer, error)
• type PushConsumer
  • func NewPushConsumer(opts ...consumer.Option) (PushConsumer, error)
• type TransactionProducer
  • func NewTransactionProducer(listener primitive.TransactionListener, opts ...producer.Option) (TransactionProducer, error)

Types

type Producer

type Producer interface {
	Start() error
	Shutdown() error
	SendSync(ctx context.Context, mq ...*primitive.Message) (*primitive.SendResult, error)
	SendAsync(ctx context.Context, mq func(ctx context.Context, result *primitive.SendResult, err error),
		msg ...*primitive.Message) error
	SendOneWay(ctx context.Context, mq ...*primitive.Message) error
	Request(ctx context.Context, ttl time.Duration, msg *primitive.Message) (*primitive.Message, error)
	RequestAsync(ctx context.Context, ttl time.Duration, callback internal.RequestCallback, msg *primitive.Message) error
}

func NewProducer

func NewProducer(opts ...producer.Option) (Producer, error)

type PullConsumer

type PullConsumer interface {
	// Start the PullConsumer for consuming message
	Start() error

	// Subscribe a topic for consuming
	Subscribe(topic string, selector consumer.MessageSelector) error

	// Unsubscribe a topic
	Unsubscribe(topic string) error

	// Shutdown the PullConsumer, all offset of MessageQueue will be commit to broker before process exit
	Shutdown() error

	// Poll messages with timeout.
	Poll(ctx context.Context, timeout time.Duration) (*consumer.ConsumeRequest, error)

	//ACK ACK
	ACK(ctx context.Context, cr *consumer.ConsumeRequest, consumeResult consumer.ConsumeResult)

	// Pull message of topic,  selector indicate which queue to pull.
	Pull(ctx context.Context, numbers int) (*primitive.PullResult, error)

	// PullFrom pull messages of queue from the offset to offset + numbers
	PullFrom(ctx context.Context, queue *primitive.MessageQueue, offset int64, numbers int) (*primitive.PullResult, error)

	// UpdateOffset updateOffset update offset of queue in mem
	UpdateOffset(queue *primitive.MessageQueue, offset int64) error

	// PersistOffset persist all offset in mem.
	PersistOffset(ctx context.Context, topic string) error

	// CurrentOffset return the current offset of queue in mem.
	CurrentOffset(queue *primitive.MessageQueue) (int64, error)
}

func NewPullConsumer

func NewPullConsumer(opts ...consumer.Option) (PullConsumer, error)

type PushConsumer

type PushConsumer interface {
	// Start the PushConsumer for consuming message
	Start() error

	// Shutdown the PushConsumer, all offset of MessageQueue will be sync to broker before process exit
	Shutdown() error
	// Subscribe a topic for consuming
	Subscribe(topic string, selector consumer.MessageSelector,
		f func(context.Context, ...*primitive.MessageExt) (consumer.ConsumeResult, error)) error

	// Unsubscribe a topic
	Unsubscribe(topic string) error

	// Suspend the consumption
	Suspend()

	// Resume the consumption
	Resume()

	// GetOffsetDiffMap Get offset difference map
	GetOffsetDiffMap() map[string]int64
}

func NewPushConsumer

func NewPushConsumer(opts ...consumer.Option) (PushConsumer, error)

type TransactionProducer

type TransactionProducer interface {
	Start() error
	Shutdown() error
	SendMessageInTransaction(ctx context.Context, mq *primitive.Message) (*primitive.TransactionSendResult, error)
}

func NewTransactionProducer

func NewTransactionProducer(listener primitive.TransactionListener, opts ...producer.Option) (TransactionProducer, error)