redis

Documentation

Overview

Package redis /**外部定义协程池方式队列

Index

• func StreamName(name string) string
• type Consumer
  • func NewConsume(rdb redis.Cmdable, name string, pool *ants.Pool, f func([]byte) error) *Consumer
  • func (r *Consumer) Consume(ctx context.Context) (topErr error)
  • func (r *Consumer) DispatchPending(expTime time.Duration, MaxLen int64)
  • func (r *Consumer) TickerPending(ctx context.Context, expTime time.Duration, perTime time.Duration, ...)
• type Pending
• type Producer
  • func NewProducer(rdb redis.Cmdable, name string, num int64) *Producer
  • func (r *Producer) Send(ctx context.Context, data map[string]interface{}) (rid string, topErr error)
  • func (r *Producer) SendSlice(ctx context.Context, data []string) (rid string, topErr error)
• type Worker
  • func (r *Worker) GetName() string
  • func (r *Worker) Init()
  • func (r *Worker) Serve(ctx context.Context) (topErr error)

Functions

func StreamName

func StreamName(name string) string

StreamName /**获取队列名称（需要加入根据预发和正式相关前缀处理）

Types

type Consumer

type Consumer struct {
	Func func(b []byte) error //接收json的处理方式
	Pool *ants.Pool
	// contains filtered or unexported fields
}

func NewConsume

func NewConsume(rdb redis.Cmdable, name string, pool *ants.Pool, f func([]byte) error) *Consumer

NewConsume 新建消费者 rdb redis客户端 name 队列名称 pool 协程池 f 处理消息方法,参数为队列消息，格式为字节数组

func (*Consumer) Consume

func (r *Consumer) Consume(ctx context.Context) (topErr error)

Consume 处理池外部定义

func (*Consumer) DispatchPending

func (r *Consumer) DispatchPending(expTime time.Duration, MaxLen int64)

DispatchPending 对超时pending进行回收 每一个pending获取一次消息（避免直接range一批因范围问题导致消息丢失）

func (*Consumer) TickerPending

func (r *Consumer) TickerPending(ctx context.Context, expTime time.Duration, perTime time.Duration, startRun bool, MaxLen int64)

TickerPending 定时任务执行pending expTime 过期时间(当时间小于等于0时设置为默认 过期时间为1天) perTime 执行周期(当时间小于等于0时设置为默认 每15分钟) startRun 是否开始就先执行一次

type Pending

type Pending struct {
	ExpTime  time.Duration //过期时间
	PerTime  time.Duration //回收触发时间间隔
	StartRun bool          //是否在开始就执行一次回收操作
	MaxLen   int64         //队列长度
}

type Producer

type Producer struct {
	MaxLen int64 //队列长度
	Approx bool  //近似长度
	// contains filtered or unexported fields
}

func NewProducer

func NewProducer(rdb redis.Cmdable, name string, num int64) *Producer

NewProducer 新建生产者 rdb redis客户端 name 队列名称 num 控制队列长度，0标示不控制

func (*Producer) Send

func (r *Producer) Send(ctx context.Context, data map[string]interface{}) (rid string, topErr error)

func (*Producer) SendSlice

func (r *Producer) SendSlice(ctx context.Context, data []string) (rid string, topErr error)

type Worker

type Worker struct {
	Name    string               //队列名称
	Handle  func(m []byte) error //处理方法
	Pending *Pending             //如果为nil则不启动pending回收
	Rdb     redis.Cmdable        //redis client
	PoolNum uint                 //线程池数，如果不传则走全局池，数目从配置文件读取
	// contains filtered or unexported fields
}

func (*Worker) GetName

func (r *Worker) GetName() string

func (*Worker) Init

func (r *Worker) Init()

func (*Worker) Serve

func (r *Worker) Serve(ctx context.Context) (topErr error)