mq

Documentation

Index

• type Consumer
• type MqOptions
  • func NewRocketmqOptions() *MqOptions
  • func (o *MqOptions) AddFlags(fs *pflag.FlagSet)
  • func (o *MqOptions) Validate() error
• type Producer
• type TransactionProducer

Types

type Consumer

type Consumer interface {
	Subscribe() error                       // 订阅消息
	ConsumerStart(...consumer.Option) error // 开启消费者Client
	ConsumerStop() error                    // 结束消费者Client
}

type MqOptions

type MqOptions struct {
	Addrs     []string `mapstructure:"addr"       json:"addr,omitempty"`
	GroupName string   `mapstructure:"group-name" json:"group-name,omitempty"`
	Topic     string   `mapstructure:"topic" json:"topic,omitempty"`
	Retry     int      `mapstructure:"retry"      json:"retry,omitempty"`
}

func NewRocketmqOptions

func NewRocketmqOptions() *MqOptions

func (*MqOptions) AddFlags

func (o *MqOptions) AddFlags(fs *pflag.FlagSet)

func (*MqOptions) Validate

func (o *MqOptions) Validate() error

type Producer

type Producer interface {
	ProducerStart() error // 开启生产者Client
	ProducerStop() error  // 结束生产者Client

	// MessageSync 同步发送（Synchronous）：这是最常见的一种发送方式，客户端发送消息后，会等待服务器端的响应。
	// 只有当消息完全被服务器接收，客户端才会继续执行。这种方式适用于对可靠性要求较高的场景。
	// shardingKeyValue用作选择特定的queue发送，如果不需要，可以为nil
	MessageSync(topic string, data []byte, shardingKeyValue *string) error

	// MessageSyncDelayTime 用同步的方式发送延迟消息
	// shardingKeyValue用作选择特定的queue发送，如果不需要，可以为nil
	MessageSyncDelayTime(topic string, data []byte, delayTimeLevel int, shardingKeyValue *string) error

	// MessageAsync 异步发送（Asynchronous）：客户端在发送消息后不会立即等待服务器的响应，而是通过一个回调接口来处理服务器的响应，
	// 这样可以提高消息发送的吞吐量。这种方式适用于对响应时间要求较敏感的场景。
	// shardingKeyValue用作选择特定的queue发送，如果不需要，可以为nil
	MessageAsync(topic string, data []byte, callback func(ctx context.Context, result *primitive.SendResult, err error), shardingKeyValue *string) error // 发送异步消息

	// MessageOneWay 单向发送（One-way）：这种方式是最轻量级的一种发送方式，客户端只负责发送消息，
	// 不等待服务器的响应，也不关心消息是否被服务器接收。这种方式的可靠性最低，但吞吐量最高，适用于对可靠性要求不高的日志收集等场景。
	// shardingKeyValue用作选择特定的queue发送，如果不需要，可以为nil
	MessageOneWay(topic string, data []byte, shardingKeyValue *string) error // 发送异步消息
}

Producer 普通消息的发送接口

type TransactionProducer

type TransactionProducer interface {
	ProducerStart(...producer.Option) error // 开启生产者Client
	ProducerStop() error

	// MessageInTransaction 发送事务消息
	MessageInTransaction(topic string, data []byte, shardingKeyValue *string) error
}

TransactionProducer 事务消息的发送接口