rmq

README

rmq

Go+Redis实现消息队列

Documentation

Index

• type ACKListMQ
  • func NewACKListMQ(client *redis.Client) *ACKListMQ
  • func (q *ACKListMQ) Consume(ctx context.Context, topic string, h Handler) error
  • func (q *ACKListMQ) SendMsg(ctx context.Context, msg *Msg) error
• type Handler
• type ListMQ
  • func NewListMQ(client *redis.Client) *ListMQ
  • func (q *ListMQ) Consume(ctx context.Context, topic string, h Handler) error
  • func (q *ListMQ) SendMsg(ctx context.Context, msg *Msg) error
• type Msg
• type PartitionACKListMQ
  • func NewPartitionACKListMQ(client *redis.Client) *PartitionACKListMQ
  • func (q *PartitionACKListMQ) Consume(ctx context.Context, topic string, partition int, h Handler) error
  • func (q *PartitionACKListMQ) SendMsg(ctx context.Context, msg *Msg) error
• type PubSubMQ
  • func NewPubSubMQ(client *redis.Client) *PubSubMQ
  • func (q *PubSubMQ) Consume(ctx context.Context, topic string, partition int, h Handler) error
  • func (q *PubSubMQ) ConsumeMultiPartitions(ctx context.Context, topic string, partitions []int, h Handler) error
  • func (q *PubSubMQ) SendMsg(ctx context.Context, msg *Msg) error
• type StreamMQ
  • func NewStreamMQ(client *redis.Client, maxLen int, approx bool) *StreamMQ
  • func (q *StreamMQ) Consume(ctx context.Context, topic, group, consumer, start string, batchSize int, ...) error
  • func (q *StreamMQ) SendMsg(ctx context.Context, msg *Msg) error

Types

type ACKListMQ

type ACKListMQ struct {
	// contains filtered or unexported fields
}

func NewACKListMQ

func NewACKListMQ(client *redis.Client) *ACKListMQ

func (*ACKListMQ) Consume

func (q *ACKListMQ) Consume(ctx context.Context, topic string, h Handler) error

Consume 返回值代表消费过程中遇到的无法处理的错误

func (*ACKListMQ) SendMsg

func (q *ACKListMQ) SendMsg(ctx context.Context, msg *Msg) error

type Handler

type Handler func(msg *Msg) error

Handler 返回值代表消息是否消费成功

type ListMQ

type ListMQ struct {
	// contains filtered or unexported fields
}

func NewListMQ

func NewListMQ(client *redis.Client) *ListMQ

func (*ListMQ) Consume

func (q *ListMQ) Consume(ctx context.Context, topic string, h Handler) error

Consume 返回值代表消费过程中遇到的无法处理的错误

func (*ListMQ) SendMsg

func (q *ListMQ) SendMsg(ctx context.Context, msg *Msg) error

type Msg

type Msg struct {
	ID        string // 消息的编号
	Topic     string // 消息的主题
	Body      []byte // 消息的Body
	Partition int    // 分区号
	Group     string // 消费者组
	Consumer  string // 消费者组里的消费者
}

Msg 消息

type PartitionACKListMQ

type PartitionACKListMQ struct {
	// contains filtered or unexported fields
}

func NewPartitionACKListMQ

func NewPartitionACKListMQ(client *redis.Client) *PartitionACKListMQ

func (*PartitionACKListMQ) Consume

func (q *PartitionACKListMQ) Consume(ctx context.Context, topic string, partition int, h Handler) error

Consume 返回值代表消费过程中遇到的无法处理的错误

func (*PartitionACKListMQ) SendMsg

func (q *PartitionACKListMQ) SendMsg(ctx context.Context, msg *Msg) error

type PubSubMQ

type PubSubMQ struct {
	// contains filtered or unexported fields
}

func NewPubSubMQ

func NewPubSubMQ(client *redis.Client) *PubSubMQ

func (*PubSubMQ) Consume

func (q *PubSubMQ) Consume(ctx context.Context, topic string, partition int, h Handler) error

Consume 返回值代表消费过程中遇到的无法处理的错误

func (*PubSubMQ) ConsumeMultiPartitions

func (q *PubSubMQ) ConsumeMultiPartitions(ctx context.Context, topic string, partitions []int, h Handler) error

ConsumeMultiPartitions 返回值代表消费过程中遇到的无法处理的错误

func (*PubSubMQ) SendMsg

func (q *PubSubMQ) SendMsg(ctx context.Context, msg *Msg) error

type StreamMQ

type StreamMQ struct {
	// contains filtered or unexported fields
}

func NewStreamMQ

func NewStreamMQ(client *redis.Client, maxLen int, approx bool) *StreamMQ

func (*StreamMQ) Consume

func (q *StreamMQ) Consume(ctx context.Context, topic, group, consumer, start string, batchSize int, h Handler) error

Consume 返回值代表消费过程中遇到的无法处理的错误 group 消费者组 consumer 消费者组里的消费者 batchSize 每次批量获取一批的大小 start 用于创建消费者组的时候指定起始消费ID，0表示从头开始消费，$表示从最后一条消息开始消费

func (*StreamMQ) SendMsg

func (q *StreamMQ) SendMsg(ctx context.Context, msg *Msg) error