nq

Documentation

Index

• func MustNewQueue(c NqConf, handler ConsumeHandler, opts ...QueueOption) queue.MessageQueue
• func NewQueue(c NqConf, handler ConsumeHandler, opts ...QueueOption) (queue.MessageQueue, error)
• type ConsumeHandle
• type ConsumeHandler
  • func WithHandle(handle ConsumeHandle) ConsumeHandler
• type MessageHandler
  • func (mh *MessageHandler) Consume(key, val string) error
• type NqConf
• type PushOption
  • func WithChunkSize(chunkSize int) PushOption
  • func WithFlushInterval(interval time.Duration) PushOption
• type Pusher
  • func NewPusher(addrs string, topic string, opts ...PushOption) *Pusher
  • func (p *Pusher) Close() error
  • func (p *Pusher) Name() string
  • func (p *Pusher) Ping() error
  • func (p *Pusher) Push(v string) error
• type QueueOption
  • func WithMetrics(metrics *stat.Metrics) QueueOption

Functions

func MustNewQueue

func MustNewQueue(c NqConf, handler ConsumeHandler, opts ...QueueOption) queue.MessageQueue

func NewQueue

func NewQueue(c NqConf, handler ConsumeHandler, opts ...QueueOption) (queue.MessageQueue, error)

Types

type ConsumeHandle

type ConsumeHandle func(key, value string) error

type ConsumeHandler

type ConsumeHandler interface {
	Consume(key, value string) error
}

func WithHandle

func WithHandle(handle ConsumeHandle) ConsumeHandler

type MessageHandler

type MessageHandler struct {
}

func (*MessageHandler) Consume

func (mh *MessageHandler) Consume(key, val string) error

type NqConf

type NqConf struct {
	// service.ServiceConf
	Brokers    []string //连接的nsqlookupd
	Group      string   //可默认为分区ID,标准版nsq无效,有赞版有用(类型为int,需要转换)
	Topic      string
	Offset     string `json:",options=first|last,default=last"` //标准版nsq无效,有赞版有用(从哪个位置读取)
	Conns      int    `json:",default=1"`                       //建立多少连接
	Consumers  int    `json:",default=1"`                       //1个连接，多个协程读取,nsq固定为1.
	Processors int    `json:",default=8"`                       //1个连接，多少协程去消费
	Channel    string `json:",default=default"`                 //消费的channel,默认为default
}

type PushOption

type PushOption func(options *chunkOptions)

func WithChunkSize

func WithChunkSize(chunkSize int) PushOption

func WithFlushInterval

func WithFlushInterval(interval time.Duration) PushOption

type Pusher

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

func NewPusher

func NewPusher(addrs string, topic string, opts ...PushOption) *Pusher

func (*Pusher) Close

func (p *Pusher) Close() error

func (*Pusher) Name

func (p *Pusher) Name() string

func (*Pusher) Ping

func (p *Pusher) Ping() error

func (*Pusher) Push

func (p *Pusher) Push(v string) error

type QueueOption

type QueueOption func(*queueOptions)

func WithMetrics

func WithMetrics(metrics *stat.Metrics) QueueOption