redis

Documentation

Index

• Constants
• type Consumer
  • func NewConsumer(configName string, config config.Config) (consumer *Consumer, err error)
  • func (consumer *Consumer) Close()
  • func (consumer *Consumer) Connect() (err error)
  • func (consumer *Consumer) Consume(task queue.TaskInfo, callback queue.ConsumeCallback) (err error)
  • func (consumer *Consumer) Start()
  • func (consumer *Consumer) Unconsume(queue string)
• type MsgBody
  • func (m *MsgBody) Body() map[string]interface{}
  • func (m *MsgBody) Header() map[string]string
  • func (m MsgBody) MarshalBinary() (data []byte, err error)
  • func (m MsgBody) String() string
• type Producer
  • func NewProducer(config config.Config, opts ...queue.Option) (m *Producer, err error)
  • func (c *Producer) Close() error
  • func (c *Producer) Count(key string) (int64, error)
  • func (c *Producer) DelayPush(key string, msg queue.Message, delaySeconds int64) error
  • func (c *Producer) Pop(key string) (string, error)
  • func (c *Producer) Push(key string, msg queue.Message) error
• type ProductOptions
• type QueueItem
  • func (qi *QueueItem) CloseMsgChan()

Constants

const (
	DELAY_QUEUE_NAME = "glue:delayqueue:list"
)

const (
	Proto = "redis"
)

Types

type Consumer

type Consumer struct {
	EnableDeadLetter bool //开启死信队列
	DeadLetterQueue  string
	// contains filtered or unexported fields
}

Consumer Consumer

func NewConsumer

func NewConsumer(configName string, config config.Config) (consumer *Consumer, err error)

NewConsumerByConfig 创建新的Consumer

func (*Consumer) Close

func (consumer *Consumer) Close()

Close 关闭当前连接

func (*Consumer) Connect

func (consumer *Consumer) Connect() (err error)

Connect 连接服务器

func (*Consumer) Consume

func (consumer *Consumer) Consume(task queue.TaskInfo, callback queue.ConsumeCallback) (err error)

Consume 注册消费信息

func (*Consumer) Start

func (consumer *Consumer) Start()

Start 启动

func (*Consumer) Unconsume

func (consumer *Consumer) Unconsume(queue string)

UnConsume 取消注册消费

type MsgBody

type MsgBody struct {
	QueueKey  string      `json:"qk,omitempty"`
	HeaderMap xtypes.SMap `json:"header,omitempty"`
	BodyMap   xtypes.XMap `json:"body,omitempty"`
	// contains filtered or unexported fields
}

func (*MsgBody) Body

func (m *MsgBody) Body() map[string]interface{}

func (*MsgBody) Header

func (m *MsgBody) Header() map[string]string

func (MsgBody) MarshalBinary

func (m MsgBody) MarshalBinary() (data []byte, err error)

func (MsgBody) String

func (m MsgBody) String() string

type Producer

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

Producer memcache配置文件

func NewProducer

func NewProducer(config config.Config, opts ...queue.Option) (m *Producer, err error)

NewProducerByConfig 根据配置文件创建一个redis连接

func (*Producer) Close

func (c *Producer) Close() error

Close 释放资源

func (*Producer) Count

func (c *Producer) Count(key string) (int64, error)

Count 获取列表中的元素个数

func (*Producer) DelayPush

func (c *Producer) DelayPush(key string, msg queue.Message, delaySeconds int64) error

Push 向存于 key 的列表的尾部插入所有指定的值

func (*Producer) Pop

func (c *Producer) Pop(key string) (string, error)

Pop 移除并且返回 key 对应的 list 的第一个元素。

func (*Producer) Push

func (c *Producer) Push(key string, msg queue.Message) error

Push 向存于 key 的列表的尾部插入所有指定的值

type ProductOptions

type ProductOptions struct {
	DelayQueueName string `json:"delay_queue_name" yaml:"delay_queue_name"`
	RangeSeconds   int    `json:"range_seconds" yaml:"range_seconds"`
	DelayInterval  int    `json:"delay_interval" yaml:"delay_interval"`
}

type QueueItem

type QueueItem struct {
	QueueName    string
	Concurrency  int //等于0 ，代表不限制
	BlockTimeout int
	// contains filtered or unexported fields
}

func (*QueueItem) CloseMsgChan

func (qi *QueueItem) CloseMsgChan()