rmqv2

Documentation

Overview

消息并发多条推送，但commit回复一次性回复，无法对单条消息进行commit回复，暂时不推荐使用 阿里云官方推荐使用 v1.2.4 版本

Index

• type Consumer
  • func NewConsumer(conf *RocketMQConfig) (c *Consumer)
  • func (c *Consumer) Close()
  • func (c *Consumer) Conn() (conn *Consumer, err error)
  • func (c *Consumer) Start() (err error)
  • func (c *Consumer) SubscribeMulti(topic, expression string, ...) (err error)
  • func (c *Consumer) SubscribeSingle(topic, expression string, ...) (err error)
  • func (c *Consumer) TopicList() (ts []string)
  • func (c *Consumer) Unsubscribe(topic string) (err error)
• type LogLevel
• type Producer
  • func NewProducer(conf *RocketMQConfig) (p *Producer)
  • func (p *Producer) Close()
  • func (p *Producer) Conn() (conn *Producer, err error)
  • func (p *Producer) SendAsyncSingle(c context.Context, message *primitive.Message) (err error)
  • func (p *Producer) SendOneWaySingle(c context.Context, message *primitive.Message) (err error)
  • func (p *Producer) SendSyncSingle(c context.Context, message *primitive.Message) (result *primitive.SendResult, err error)
• type RocketMQConfig

Types

type Consumer

type Consumer struct {
	Consumer rocketmq.PushConsumer
	// contains filtered or unexported fields
}

func NewConsumer

func NewConsumer(conf *RocketMQConfig) (c *Consumer)

new

func (*Consumer) Close

func (c *Consumer) Close()

unsubscribe all topic

func (*Consumer) Conn

func (c *Consumer) Conn() (conn *Consumer, err error)

connect to aliyun rocketmq

func (*Consumer) Start

func (c *Consumer) Start() (err error)

start subscribe

func (*Consumer) SubscribeMulti

func (c *Consumer) SubscribeMulti(topic, expression string, callback func(ctx context.Context, ext ...*primitive.MessageExt) error) (err error)

多条消息消费，需配置 client.MessageBatchMaxSize() 且size不为 1，否则不生效

func (*Consumer) SubscribeSingle

func (c *Consumer) SubscribeSingle(topic, expression string, callback func(ctx context.Context, ext *primitive.MessageExt) error) (err error)

单条消息消费 default

func (*Consumer) TopicList

func (c *Consumer) TopicList() (ts []string)

get topic list

func (*Consumer) Unsubscribe

func (c *Consumer) Unsubscribe(topic string) (err error)

unsubscribe one topic

type LogLevel

type LogLevel string

const (
	LogDebug LogLevel = "debug"
	LogWarn  LogLevel = "warn"
	LogError LogLevel = "error"
	LogInfo  LogLevel = "info"
)

type Producer

type Producer struct {
	Producer rocketmq.Producer
	// contains filtered or unexported fields
}

func NewProducer

func NewProducer(conf *RocketMQConfig) (p *Producer)

func (*Producer) Close

func (p *Producer) Close()

func (*Producer) Conn

func (p *Producer) Conn() (conn *Producer, err error)

connect to aliyun rocketmq

func (*Producer) SendAsyncSingle

func (p *Producer) SendAsyncSingle(c context.Context, message *primitive.Message) (err error)

异步单条消息发送，对应消费 topic 的 MessageBatchMaxSize = 1时用

func (*Producer) SendOneWaySingle

func (p *Producer) SendOneWaySingle(c context.Context, message *primitive.Message) (err error)

func (*Producer) SendSyncSingle

func (p *Producer) SendSyncSingle(c context.Context, message *primitive.Message) (result *primitive.SendResult, err error)

同步单条消息发送，对应消费 topic 的 MessageBatchMaxSize = 1时用

type RocketMQConfig

type RocketMQConfig struct {
	// 阿里云 实例ID
	Namespace string
	// GroupID 阿里云创建
	GroupName string
	// 设置 TCP 协议接入点，从阿里云 RocketMQ 控制台的实例详情页面获取。
	EndPoint string
	// 您在阿里云账号管理控制台中创建的 AccessKeyId，用于身份认证。
	AccessKey string
	// 您在阿里云账号管理控制台中创建的 AccessKeySecret，用于身份认证。
	SecretKey string
	// log 级别 // default info
	LogLevel LogLevel
	// 自定义消费者配置
	ConsumerOptions []consumer.Option
	// 自定义生产者配置
	ProducerOptions []producer.Option
}