rocket_mq

Documentation

Index

• Constants
• type Config
• type Consumer
  • func (consumer *Consumer) Shutdown() error
  • func (consumer *Consumer) Start() error
• type ConsumerConfig
• type ConsumerHandler
• type Message
  • func (m *Message) GetMessageType(messageTypeName string) MessageType
  • func (m *Message) IsLegalTopic() error
  • func (m *Message) IsLegalType() error
• type MessageType
  • func (messageType MessageType) ToString() string
• type Producer
  • func (producer *Producer) Push(ctx context.Context, msgPackage Message) error
  • func (producer *Producer) Shutdown() error
  • func (producer *Producer) Start() error
• type ProducerConfig
• type RocketMq
  • func NewClient(cfg Config) *RocketMq
  • func (mq *RocketMq) Id() string
  • func (mq *RocketMq) NewConsumer(cfg ConsumerConfig) *Consumer
  • func (mq *RocketMq) NewProducer(cfg *ProducerConfig) *Producer
  • func (mq *RocketMq) Running()
  • func (mq *RocketMq) Stop() error
• type TransactionCheckFunc

Constants

const (
	MsgTypeNormalName      = "Normal"      // 普通消息
	MsgTypeFIFOName        = "FIFO"        // 顺序消息
	MsgTypeDelayName       = "Delay"       // 延迟消息
	MsgTypeTransactionName = "Transaction" // 事务消息
)

Types

type Config

type Config struct {
	EndPoint     string        `json:"endPoint" yaml:"endPoint"`
	Group        string        `json:"group" yaml:"group"`
	Region       string        `json:"region" yaml:"region"`
	AccessKey    string        `json:"accessKey" yaml:"accessKey"`
	AccessSecret string        `json:"accessSecret" yaml:"accessSecret"`
	DialTimeout  time.Duration `json:"dialTimeout" yaml:"dialTimeout"`
}

type Consumer

type Consumer struct {
	Id      string
	Name    string
	Handler ConsumerHandler
	// contains filtered or unexported fields
}

func (*Consumer) Shutdown

func (consumer *Consumer) Shutdown() error

func (*Consumer) Start

func (consumer *Consumer) Start() error

type ConsumerConfig

type ConsumerConfig struct {
	Name              string
	ConsumerGroup     string
	Topics            []string
	DialDuration      time.Duration
	InvisibleDuration time.Duration
	MaxMessageNum     int
	Handler           ConsumerHandler
}

type ConsumerHandler

type ConsumerHandler func(ctx context.Context, message Message) error

type Message

type Message struct {
	RequestId     string
	MsgId         string
	UserMessageId string
	MessageType   MessageType
	Body          any
	Topic         string
	MessageGroup  string
	DelayTime     time.Duration
}

func (*Message) GetMessageType

func (m *Message) GetMessageType(messageTypeName string) MessageType

func (*Message) IsLegalTopic

func (m *Message) IsLegalTopic() error

func (*Message) IsLegalType

func (m *Message) IsLegalType() error

type MessageType

type MessageType int

const (
	MsgTypeNormal      MessageType = 8 << iota // 普通消息
	MsgTypeFIFO                                // 顺序消息
	MsgTypeDelay                               // 延迟消息
	MsgTypeTransaction                         // 事务消息
)

消息类型的常量定义

func (MessageType) ToString

func (messageType MessageType) ToString() string

type Producer

type Producer struct {
	Id   string
	Name string
	// contains filtered or unexported fields
}

func (*Producer) Push

func (producer *Producer) Push(ctx context.Context, msgPackage Message) error

func (*Producer) Shutdown

func (producer *Producer) Shutdown() error

func (*Producer) Start

func (producer *Producer) Start() error

type ProducerConfig

type ProducerConfig struct {
	Name               string
	Topics             []string
	TransactionChecker TransactionCheckFunc
}

type RocketMq

type RocketMq struct {
	ProducerList map[string]*Producer
	ConsumerList map[string]*Consumer
	// contains filtered or unexported fields
}

func NewClient

func NewClient(cfg Config) *RocketMq

func (*RocketMq) Id

func (mq *RocketMq) Id() string

func (*RocketMq) NewConsumer

func (mq *RocketMq) NewConsumer(cfg ConsumerConfig) *Consumer

func (*RocketMq) NewProducer

func (mq *RocketMq) NewProducer(cfg *ProducerConfig) *Producer

func (*RocketMq) Running

func (mq *RocketMq) Running()

func (*RocketMq) Stop

func (mq *RocketMq) Stop() error

type TransactionCheckFunc

type TransactionCheckFunc func(msg *golang.MessageView) golang.TransactionResolution