rmq

Documentation

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, ...) (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()

Close unsubscribe all topic

func (*Consumer) Conn

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

Conn connect to aliyun rocketmq

func (*Consumer) Start

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

Start start subscribe

func (*Consumer) SubscribeMulti

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

SubscribeMulti 多条消息消费，需配置 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)

SubscribeSingle 单条消息消费 default

func (*Consumer) TopicList

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

TopicList get topic list

func (*Consumer) Unsubscribe

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

Unsubscribe 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)

Conn connect to aliyun rocketmq

func (*Producer) SendAsyncSingle

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

SendAsyncSingle 异步单条消息发送，对应消费 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)

SendSyncSingle 同步单条消息发送，对应消费 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
	// currently consume limiter
	Limit *limit.Config
}