disruptor

Documentation

Index

• type Consumer
  • func NewConsumer(opt *ConsumerOptions, rdsCli redis.UniversalClient) (Consumer, error)
• type ConsumerOptions
• type ErrorNotifier
• type Handler
• type Marshaler
• type Message
• type Producer
  • func NewProducer(opt *ProducerOptions, rdsCli redis.UniversalClient) (Producer, error)
• type ProducerOptions

Types

type Consumer

type Consumer interface {
	Pop(data Marshaler, h Handler) (error, bool)
	Close()
}

func NewConsumer

func NewConsumer(opt *ConsumerOptions, rdsCli redis.UniversalClient) (Consumer, error)

type ConsumerOptions

type ConsumerOptions struct {
	QueueName         string
	Consumer          string
	ShardsCount       int8
	PrefetchCount     int64         // 每次从队列中读取的消息数量
	Block             time.Duration // 读取队列数据时阻塞的时长
	PendingBufferSize int64         // 本地缓冲队列长度
	PipeBufferSize    int64         // 每次批量ack的数量
	PipePeriod        time.Duration // 每次ack的时间间隔
	ErrorNotifier     ErrorNotifier
}

type ErrorNotifier

type ErrorNotifier interface {
	EmitError(error)
}

type Handler

type Handler func(m Message) error

type Marshaler

type Marshaler interface {
	Marshal() ([]byte, error)
	Unmarshal(data []byte) error
}

type Message

type Message struct {
	ID     string
	Stream string
	Group  string
	Body   []byte
}

Message from queue

type Producer

type Producer interface {
	Close()
	Push(data Marshaler) error
}

func NewProducer

func NewProducer(opt *ProducerOptions, rdsCli redis.UniversalClient) (Producer, error)

type ProducerOptions

type ProducerOptions struct {
	QueueName         string
	ShardsCount       int8
	PendingBufferSize int64         // 本地消息缓冲的大小
	PipeBufferSize    int64         // 每次批量发送的数量
	PipePeriod        time.Duration // 批量发送数据的时间间隔
	ErrorNotifier     ErrorNotifier
}