rocketmq

Documentation

Index

• type Config
  • func DefaultConfig() *Config
  • func RawConfig(key string) *Config
• type ConsumerConfig
  • func DefaultConsumerConfig() *ConsumerConfig
  • func RawConsumerConfig(key string) *ConsumerConfig
  • func StdPushConsumerConfig(name string) *ConsumerConfig
  • func (conf *ConsumerConfig) Build() *PushConsumer
• type FlowInfo
• type Producer
  • func StdNewProducer(name string) *Producer
  • func (pc *Producer) Close() error
  • func (pc *Producer) Send(msg []byte) error
  • func (pc *Producer) SendMsg(msg *primitive.Message) (*primitive.SendResult, error)
  • func (pc *Producer) SendWithContext(ctx context.Context, msg []byte) error
  • func (pc *Producer) SendWithMsg(ctx context.Context, msg *primitive.Message) error
  • func (pc *Producer) SendWithResult(msg []byte, tag string) (*primitive.SendResult, error)
  • func (pc *Producer) SendWithTag(msg []byte, tag string) error
  • func (pc *Producer) Start() error
  • func (pc *Producer) WithInterceptor(fs ...primitive.Interceptor) *Producer
• type ProducerConfig
  • func DefaultProducerConfig() *ProducerConfig
  • func RawProducerConfig(key string) *ProducerConfig
  • func StdProducerConfig(name string) *ProducerConfig
  • func (conf *ProducerConfig) Build() *Producer
• type PushConsumer
  • func (cc *PushConsumer) Close()
  • func (cc *PushConsumer) RegisterBatchMessage(f func(context.Context, ...*primitive.MessageExt) error) *PushConsumer
  • func (cc *PushConsumer) RegisterSingleMessage(f func(context.Context, *primitive.MessageExt) error) *PushConsumer
  • func (cc *PushConsumer) Start() error
  • func (cc *PushConsumer) Subscribe(topic string, f func(context.Context, *primitive.MessageExt) error) *PushConsumer
  • func (cc *PushConsumer) WithInterceptor(fs ...primitive.Interceptor) *PushConsumer
• type Shadow

Types

type Config

type Config struct {
	Addresses []string        `json:"addr" toml:"addr"`
	Consumer  *ConsumerConfig `json:"consumer" toml:"consumer"`
	Producer  *ProducerConfig `json:"producer" toml:"producer"`
}

Config config...

func DefaultConfig

func DefaultConfig() *Config

DefaultConfig ...

func RawConfig

func RawConfig(key string) *Config

RawConfig 返回配置

type ConsumerConfig

type ConsumerConfig struct {
	Name            string        `json:"name" toml:"name"`
	Enable          bool          `json:"enable" toml:"enable"`
	Addr            []string      `json:"addr" toml:"addr"`
	Topic           string        `json:"topic" toml:"topic"`
	Group           string        `json:"group" toml:"group"`
	DialTimeout     time.Duration `json:"dialTimeout" toml:"dialTimeout"`
	RwTimeout       time.Duration `json:"rwTimeout" toml:"rwTimeout"`
	SubExpression   string        `json:"subExpression" toml:"subExpression"`
	Rate            float64       `json:"rate" toml:"rate"`
	Capacity        int64         `json:"capacity" toml:"capacity"`
	WaitMaxDuration time.Duration `json:"waitMaxDuration" toml:"waitMaxDuration"`
	Shadow          Shadow        `json:"shadow" toml:"shadow"`
	Reconsume       int32         `json:"reconsume" toml:"reconsume"`
	AccessKey       string        `json:"accessKey" toml:"accessKey"`
	SecretKey       string        `json:"secretKey" toml:"secretKey"`
	MessageModel    string        `json:"messageModel" toml:"messageModel"` // 消费模式,默认clustering
	// client实例名，默认会基于Addr字段生成md5，支持多集群
	InstanceName string `json:"instanceName" toml:"instanceName"`
	EnableTrace  bool   `json:"enableTrace" toml:"enableTrace"`
	// 批量消费的最大消息数量，取值范围：[1, 1024]，默认值为1
	ConsumeMessageBatchMaxSize int `json:"consumeMessageBatchMaxSize" toml:"consumeMessageBatchMaxSize"`
	// 每批次从broker拉取消息的最大个数，取值范围：[1, 1024]，默认值为32
	PullBatchSize int32 `json:"pullBatchSize" toml:"pullBatchSize"`
}

ConsumerConfig consumer config

func DefaultConsumerConfig

func DefaultConsumerConfig() *ConsumerConfig

DefaultConsumerConfig ...

func RawConsumerConfig

func RawConsumerConfig(key string) *ConsumerConfig

RawConsumerConfig 返回配置

func StdPushConsumerConfig

func StdPushConsumerConfig(name string) *ConsumerConfig

StdPushConsumerConfig ...

func (*ConsumerConfig) Build

func (conf *ConsumerConfig) Build() *PushConsumer

type FlowInfo

type FlowInfo struct {
	Name      string   `json:"name"`
	Addr      []string `json:"addr"`
	Topic     string   `json:"topic"`
	Group     string   `json:"group"`
	GroupType string   `json:"groupType"` // 类型， consumer 消费者， producer 生产者
	istats.FlowInfoBase
}

type Producer

type Producer struct {
	rocketmq.Producer

	ProducerConfig
	// contains filtered or unexported fields
}

func StdNewProducer

func StdNewProducer(name string) *Producer

func (*Producer) Close

func (pc *Producer) Close() error

func (*Producer) Send

func (pc *Producer) Send(msg []byte) error

Send rocketmq发送消息

func (*Producer) SendMsg

func (pc *Producer) SendMsg(msg *primitive.Message) (*primitive.SendResult, error)

SendMsg... 自定义消息格式

func (*Producer) SendWithContext

func (pc *Producer) SendWithContext(ctx context.Context, msg []byte) error

SendWithContext 发送消息

func (*Producer) SendWithMsg

func (pc *Producer) SendWithMsg(ctx context.Context, msg *primitive.Message) error

SendWithMsg... 自定义消息格式

func (*Producer) SendWithResult

func (pc *Producer) SendWithResult(msg []byte, tag string) (*primitive.SendResult, error)

SendWithResult rocket mq 发送消息,可以自定义选择 tag 及返回结果

func (*Producer) SendWithTag

func (pc *Producer) SendWithTag(msg []byte, tag string) error

SendWithTag rocket mq 发送消息,可以自定义选择 tag

func (*Producer) Start

func (pc *Producer) Start() error

func (*Producer) WithInterceptor

func (pc *Producer) WithInterceptor(fs ...primitive.Interceptor) *Producer

type ProducerConfig

type ProducerConfig struct {
	Name        string        `json:"name" toml:"name"`
	Addr        []string      `json:"addr" toml:"addr"`
	Topic       string        `json:"topic" toml:"topic"`
	Group       string        `json:"group" toml:"group"`
	Retry       int           `json:"retry" toml:"retry"`
	DialTimeout time.Duration `json:"dialTimeout" toml:"dialTimeout"`
	RwTimeout   time.Duration `json:"rwTimeout" toml:"rwTimeout"`
	Shadow      Shadow        `json:"shadow" toml:"shadow"`
	AccessKey   string        `json:"accessKey" toml:"accessKey"`
	SecretKey   string        `json:"secretKey" toml:"secretKey"`
	// client实例名，默认会基于Addr字段生成md5，支持多集群
	InstanceName string `json:"instanceName" toml:"instanceName"`
	EnableTrace  bool   `json:"enableTrace" toml:"enableTrace"`
}

ProducerConfig producer config

func DefaultProducerConfig

func DefaultProducerConfig() *ProducerConfig

DefaultProducerConfig ...

func RawProducerConfig

func RawProducerConfig(key string) *ProducerConfig

RawProducerConfig ...

func StdProducerConfig

func StdProducerConfig(name string) *ProducerConfig

StdProducerConfig ...

func (*ProducerConfig) Build

func (conf *ProducerConfig) Build() *Producer

type PushConsumer

type PushConsumer struct {
	rocketmq.PushConsumer

	ConsumerConfig
	// contains filtered or unexported fields
}

func (*PushConsumer) Close

func (cc *PushConsumer) Close()

func (*PushConsumer) RegisterBatchMessage

func (cc *PushConsumer) RegisterBatchMessage(f func(context.Context, ...*primitive.MessageExt) error) *PushConsumer

func (*PushConsumer) RegisterSingleMessage

func (cc *PushConsumer) RegisterSingleMessage(f func(context.Context, *primitive.MessageExt) error) *PushConsumer

func (*PushConsumer) Start

func (cc *PushConsumer) Start() error

func (*PushConsumer) Subscribe

func (cc *PushConsumer) Subscribe(topic string, f func(context.Context, *primitive.MessageExt) error) *PushConsumer

Deprecated: use RegisterSingleMessage or RegisterBatchMessage instead

func (*PushConsumer) WithInterceptor

func (cc *PushConsumer) WithInterceptor(fs ...primitive.Interceptor) *PushConsumer

type Shadow

type Shadow struct {
	Mode string `json:"mode" toml:"mode"`
	// mode开启模式下白名单内topic不进行丢弃
	WitheTopics []string `json:"witheTopics" toml:"witheTopics"`
}