rmqv2

Documentation

Overview

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

Index

• type Consumer
  • func NewConsumer(conf *RocketMQConfig) (c *Consumer)
  • func (c *Consumer) ConsumeMessageBatchMaxSize(size int) *Consumer
  • func (c *Consumer) MessageModel(messageModel consumer.MessageModel) *Consumer
  • func (c *Consumer) PullBatchSize(size int) *Consumer
  • func (c *Consumer) Shutdown() (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) Unsubscribe(topic string) (err error)
• type LogLevel
• type Producer
  • func NewProducer(conf *RocketMQConfig) (p *Producer)
  • func (p *Producer) SendSyncMulti(c context.Context, messages []*primitive.Message) (result *primitive.SendResult, err error)
  • func (p *Producer) SendSyncSingle(c context.Context, message *primitive.Message) (result *primitive.SendResult, err error)
  • func (p *Producer) Shutdown() (err error)
  • func (p *Producer) Start() (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)

func (*Consumer) ConsumeMessageBatchMaxSize

func (c *Consumer) ConsumeMessageBatchMaxSize(size int) *Consumer

func (*Consumer) MessageModel

func (c *Consumer) MessageModel(messageModel consumer.MessageModel) *Consumer

func (*Consumer) PullBatchSize

func (c *Consumer) PullBatchSize(size int) *Consumer

func (*Consumer) Shutdown

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

func (*Consumer) Start

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

func (*Consumer) SubscribeMulti

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

多条消息消费，需配置 client.ConsumeMessageBatchMaxSize() 且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) Unsubscribe

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

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

func (p *Producer) SendSyncMulti(c context.Context, messages []*primitive.Message) (result *primitive.SendResult, err error)

func (*Producer) SendSyncSingle

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

func (*Producer) Shutdown

func (p *Producer) Shutdown() (err error)

func (*Producer) Start

func (p *Producer) Start() (err error)

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
}