queue

Documentation

Overview

Package queue 队列对象 非常适合作为简单的生产者消费者模式的中间件

Package queue 队列对象 非常适合作为简单的生产者消费者模式的中间件

Package queue 队列对象 非常适合作为简单的生产者消费者模式的中间件

Index

• Variables
• type Consumer
  • func NewConsumer(kb *clientkeybatch.ClientKeyBatch, opts ...broker.Option) *Consumer
  • func (s *Consumer) AsQueueArray() []*Queue
  • func (s *Consumer) Get(ctx context.Context, timeout time.Duration) (string, string, error)
  • func (s *Consumer) Listen(asyncHanddler bool, p ...event.Parser) error
  • func (s *Consumer) StopListening() error
• type Producer
  • func NewProducer(k *clientkey.ClientKey, opts ...broker.Option) *Producer
  • func (p *Producer) AsQueue() *Queue
  • func (p *Producer) PubEvent(ctx context.Context, payload interface{}) (*event.Event, error)
  • func (p *Producer) Publish(ctx context.Context, payload interface{}) error
• type Queue
  • func New(k *clientkey.ClientKey) *Queue
  • func (q *Queue) AsProducer(opts ...broker.Option) *Producer
  • func (q *Queue) Len(ctx context.Context) (int64, error)

Variables

var ErrQueueAlreadyListened = errors.New("队列已经被监听了")

ErrQueueAlreadyListened 队列已经被监听了

var ErrQueueNotListeningYet = errors.New("队列未被监听")

ErrQueueNotListeningYet 队列未被监听

var ErrQueueResNotTwo = errors.New("从队列中得到的消息结果不为2位")

ErrQueueResNotTwo 从队列中得到的消息结果不为2位

Types

type Consumer

type Consumer struct {
	*clientkeybatch.ClientKeyBatch
	*consumerabc.ConsumerABC
	// contains filtered or unexported fields
}

Consumer 队列消费者对象

func NewConsumer

func NewConsumer(kb *clientkeybatch.ClientKeyBatch, opts ...broker.Option) *Consumer

NewConsumer 创建一个新的队列消费者对象 @params k *clientkeybatch.ClientKeyBatch redis客户端的批键对象 @params opts ...broker.Option 消费者的配置

func (*Consumer) AsQueueArray

func (s *Consumer) AsQueueArray() []*Queue

AsQueueArray 从消费者构造由队列对象组成的序列 序列顺序与keys中键的顺序一致

func (*Consumer) Get

func (s *Consumer) Get(ctx context.Context, timeout time.Duration) (string, string, error)

Get 从多个队列中取出数据,timeout为0则表示一直阻塞直到有数据 @params ctx context.Context 请求的上下文 @params timeout time.Duration 等待超时时间 @returns string, string, error 依顺序为topic,payload,err

func (*Consumer) Listen

func (s *Consumer) Listen(asyncHanddler bool, p ...event.Parser) error

Listen 监听一个队列 @params asyncHanddler bool 是否并行执行回调 @params p ...Parser 监听队列的解析函数

func (*Consumer) StopListening

func (s *Consumer) StopListening() error

StopListening 停止监听

type Producer

type Producer struct {
	*clientkey.ClientKey
	// contains filtered or unexported fields
}

Producer 队列的生产者对象

func NewProducer

func NewProducer(k *clientkey.ClientKey, opts ...broker.Option) *Producer

NewProducer 创建一个新的队列生产者对象 @params k *clientkey.ClientKey redis客户端的键对象 @params opts ...broker.Option 生产者的配置

func (*Producer) AsQueue

func (p *Producer) AsQueue() *Queue

AsQueue 由生产构造队列对象

func (*Producer) PubEvent

func (p *Producer) PubEvent(ctx context.Context, payload interface{}) (*event.Event, error)

PubEvent 向队列中放入事件数据 @params ctx context.Context 请求的上下文 @params payload []byte 发送的消息负载 @returns *event.Event 发送出去的消息对象

func (*Producer) Publish

func (p *Producer) Publish(ctx context.Context, payload interface{}) error

Publish 向队列中放入数据 @params ctx context.Context 请求的上下文 @params payload interface{} 发送的消息负载,负载支持string,bytes,bool,number,以及可以被json或者msgpack序列化的对象

type Queue

type Queue struct {
	*clientkey.ClientKey
}

Queue 消息队列

func New

func New(k *clientkey.ClientKey) *Queue

New 创建一个新的队列对象 @params k *clientkey.ClientKey redis客户端的键对象

func (*Queue) AsProducer

func (q *Queue) AsProducer(opts ...broker.Option) *Producer

func (*Queue) Len

func (q *Queue) Len(ctx context.Context) (int64, error)

Len 查看当前队列长度 @params ctx context.Context 请求的上下文