queuehelper

Documentation

Overview

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

Package 双端队列对象

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

queuehelper 满足pchelper规定的生产者和消费者接口

Index

• Variables
• func SerializeWithJSON() optparams.Option[Options]
• func SerializeWithMsgpack() optparams.Option[Options]
• func WithBlockTime(d time.Duration) optparams.Option[Options]
• func WithClientID(clientID string) optparams.Option[Options]
• func WithUUIDSnowflake() optparams.Option[Options]
• func WithUUIDSonyflake() optparams.Option[Options]
• func WithUUIDv4() optparams.Option[Options]
• type Consumer
  • func NewConsumer(cli redis.UniversalClient, opts ...optparams.Option[Options]) (*Consumer, error)
  • func (s *Consumer) Client() redis.UniversalClient
  • func (s *Consumer) Get(ctx context.Context, timeout time.Duration, topics ...string) (string, string, error)
  • func (s *Consumer) Len(ctx context.Context, topic string) (int64, error)
  • func (s *Consumer) Listen(topics string, opts ...optparams.Option[pchelper.ListenOptions]) error
  • func (s *Consumer) StopListening() error
• type Options
• type Producer
  • func NewProducer(cli redis.UniversalClient, opts ...optparams.Option[Options]) (*Producer, error)
  • func (p *Producer) Client() redis.UniversalClient
  • func (p *Producer) Len(ctx context.Context, topic string) (int64, error)
  • func (p *Producer) PubEvent(ctx context.Context, topic string, payload interface{}, ...) (*pchelper.Event, error)
  • func (p *Producer) Publish(ctx context.Context, topic string, payload interface{}, ...) error

Variables

var ErrQueueAlreadyListened = errors.New("queue already listened")

ErrQueueAlreadyListened 队列已经被监听了

var ErrQueueNotListeningYet = errors.New("queue not listening yet")

ErrQueueNotListeningYet 队列未被监听

var ErrQueueResNotTwo = errors.New("queue result not 2")

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

Functions

func SerializeWithJSON

func SerializeWithJSON() optparams.Option[Options]

SerializeWithJSON 使用JSON作为序列化反序列化的协议

func SerializeWithMsgpack

func SerializeWithMsgpack() optparams.Option[Options]

SerializeWithMsgpack 使用JSON作为序列化反序列化的协议

func WithBlockTime

func WithBlockTime(d time.Duration) optparams.Option[Options]

WithBlockTime 设置客户端阻塞等待消息的时长

func WithClientID

func WithClientID(clientID string) optparams.Option[Options]

WithClientID 中间件通用设置,设置客户端id

func WithUUIDSnowflake

func WithUUIDSnowflake() optparams.Option[Options]

WithUUIDSnowflake 使用snowflake作为uuid的生成器

func WithUUIDSonyflake

func WithUUIDSonyflake() optparams.Option[Options]

WithUUIDSonyflake 使用sonyflake作为uuid的生成器

func WithUUIDv4

func WithUUIDv4() optparams.Option[Options]

WithUUIDv4 使用uuid4作为uuid的生成器

Types

type Consumer

type Consumer struct {
	*clientIdhelper.ClientIDAbc
	*pchelper.ConsumerABC
	// contains filtered or unexported fields
}

Consumer 队列消费者对象

func NewConsumer

func NewConsumer(cli redis.UniversalClient, opts ...optparams.Option[Options]) (*Consumer, error)

NewConsumer 创建一个新的队列消费者对象 @params cli redis.UniversalClient redis客户端对象 @params opts ...optparams.Option[Options] 消费者的配置

func (*Consumer) Client

func (s *Consumer) Client() redis.UniversalClient

Client 获取连接的redis客户端

func (*Consumer) Get

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

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

func (*Consumer) Len

func (s *Consumer) Len(ctx context.Context, topic string) (int64, error)

Len 查看当前队列长度 @params ctx context.Context 请求的上下文 @params topic string 指定要查看的队列名

func (*Consumer) Listen

func (s *Consumer) Listen(topics string, opts ...optparams.Option[pchelper.ListenOptions]) error

Listen 监听一个队列 @params topics string 监听的topic,复数topic用`,`隔开 @params opts ...optparams.Option[pchelper.ListenOptions] 监听时的一些配置,具体看listenoption.go说明

func (*Consumer) StopListening

func (s *Consumer) StopListening() error

StopListening 停止监听

type Options

type Options struct {
	BlockTime            time.Duration                              //queue结构使用的参数,用于设置每次拉取的阻塞时长
	ProducerConsumerOpts []optparams.Option[pchelper.Options]       //初始化pchelper的配置
	ClientIDOpts         []optparams.Option[clientIdhelper.Options] //初始化ClientID的配置
}

type Producer

type Producer struct {
	*pchelper.ProducerConsumerABC
	*clientIdhelper.ClientIDAbc
	// contains filtered or unexported fields
}

Producer 队列的生产者对象

func NewProducer

func NewProducer(cli redis.UniversalClient, opts ...optparams.Option[Options]) (*Producer, error)

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

func (*Producer) Client

func (p *Producer) Client() redis.UniversalClient

Client 获取连接的redis客户端

func (*Producer) Len

func (p *Producer) Len(ctx context.Context, topic string) (int64, error)

Len 查看当前队列长度 @params ctx context.Context 请求的上下文 @params topic string 指定要查看的队列名

func (*Producer) PubEvent

func (p *Producer) PubEvent(ctx context.Context, topic string, payload interface{}, opts ...optparams.Option[pchelper.PublishOptions]) (*pchelper.Event, error)

PubEvent 向队列中放入事件数据 @params ctx context.Context 请求的上下文 @params topic string 发送去的指定频道 @params payload []byte 发送的消息负载 @params opts ...optparams.Option[pchelper.PublishOptions] 无效 @returns *pchelper.Event 发送出去的消息对象

func (*Producer) Publish

func (p *Producer) Publish(ctx context.Context, topic string, payload interface{}, opts ...optparams.Option[pchelper.PublishOptions]) error

Publish 向队列中放入数据 @params ctx context.Context 请求的上下文 @params topic string 发送去的指定双端队列 @params payload interface{} 发送的消息负载,负载支持string,bytes,bool,number,以及可以被json或者msgpack序列化的对象 @params opts ...optparams.Option[pchelper.PublishOptions] 无效