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 GetProducer(name string) *Producer
  • func InvokerProducer(name string) *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) 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 GetConsumer(name string) *PushConsumer
  • func (cc *PushConsumer) Close() error
  • 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"`
}

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 GetProducer

func GetProducer(name string) *Producer

func InvokerProducer

func InvokerProducer(name string) *Producer

Invoker ...

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) 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"`
}

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 GetConsumer

func GetConsumer(name string) *PushConsumer

Get ...

func (*PushConsumer) Close

func (cc *PushConsumer) Close() error

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

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"`
}