go_delay_queue

README

go-delay-queue

WIKI

https://github.com/wsqun/go-delay-queue/wiki/Introduction

Installation

make sure to initialize a Go module before installing go-delay-queue:

go get https://github.com/wsqun/go-delay-queue

Import

import "github.com/wsqun/go-delay-queue"

QuickStart

demo操作

kafka可使用docker搭建

Documentation

Index

• type DelayLevel
• type DelayServeConf
• type DelayTopicMsg
• type Delayer
  • func NewDelay(conf *DelayServeConf, queuer Iqueue, opt ...OptFn) (dr *Delayer, err error)
  • func (dr *Delayer) AddMsg(level int, msg interface{}) (err error)
  • func (dr *Delayer) Run()
  • func (dr *Delayer) SetDurationMin(ttl time.Duration)
• type Iqueue
• type OptFn

Types

type DelayLevel

type DelayLevel struct {
	TopicName string                    //队列名
	Level     int                       // 延迟等级
	RetryNums int                       // 重试次数
	NoAlive   bool                      // 当为true 不获取延迟消息
	Ttl       time.Duration             // 延迟时间
	DealFn    func(DelayTopicMsg) error // 处理方法
}

延迟等级定义

type DelayServeConf

type DelayServeConf struct {
	ClientCtx   context.Context
	ClientWg    *sync.WaitGroup
	DelayLevels []*DelayLevel
	Debug       bool
}

配置

type DelayTopicMsg

type DelayTopicMsg struct {
	Id        uint64      `json:"id" from:"id"`
	Level     int         `json:"level" from:"level"`           // 延迟等级
	ExpiredAt int64       `json:"expired_at" from:"expired_at"` // 生效时间
	RetryNums int         `json:"retry_nums" from:"retry_nums"` // 重试次数
	DelayMsg  interface{} `json:"delay_msg" from:"delay_msg"`   // 延迟消息
}

队列延迟消息

type Delayer

type Delayer struct {
	Debug bool
	// contains filtered or unexported fields
}

func NewDelay

func NewDelay(conf *DelayServeConf, queuer Iqueue, opt ...OptFn) (dr *Delayer, err error)

func (*Delayer) AddMsg

func (dr *Delayer) AddMsg(level int, msg interface{}) (err error)

添加延迟消息

func (*Delayer) Run

func (dr *Delayer) Run()

func (*Delayer) SetDurationMin

func (dr *Delayer) SetDurationMin(ttl time.Duration)

调整最小延迟时间

type Iqueue

type Iqueue interface {
	// 订阅消息
	// topic 为队列名，dealFn为消息处理逻辑方法，[]byte为队列消息，当err发生则放回消息，默认为程序退出
	SubscribeMsg(topic string, dealFn func([]byte) (err error)) (err error)
	// 生产消息
	PublishMsg(topic string, msg []byte) (err error)
}

type OptFn

type OptFn func(dr *Delayer)