delayqueue

README

go-delayqueue

基于redis zset 实现延迟队列

• http服务查看队列详情
• 集群支持
• 支持批量消费消息 TODO
• 重试机制
• ACK确认机制 TODO

消费者

func myJob(msg interface{}) {
	fmt.Println(msg)
    // 业务重试
    delayqueue.RetryEnqueue("myJob", msg)
}

func main() {
    go delayqueue.StatsServer(8000)

	delayqueue.Configure(map[string]string{
		"addr":  "localhost:6379",
		"namespace": "myDelay",
	})
	delayqueue.Process("myJob", myJob, 10)
	delayqueue.Run()
}

生产者

func main() {
	delayqueue.Configure(map[string]string{
		"addr":      "localhost:6379",
		"namespace": "myDelay",
	})
	delayqueue.Enqueue("myJob", map[string]string{
		"content": "hello world",
	}, 3*time.Second)
}

Documentation

Index

• Constants
• func Enqueue(queue string, message interface{}, delayTime time.Duration) (string, error)
• func GetGoroutineID() uint64
• func InitClient(opt *redis.Options)
• func InitClusterClient(opt *redis.ClusterOptions)
• func Process(queue string, delayFunc delayFunc, concurrency int, params ...Params)
• func RetryEnqueue(queue string, member interface{}) (string, error)
• func Run()
• func Stats(w http.ResponseWriter, req *http.Request)
• func StatsServer(port int)
• type EnqueueData
• type Params

Constants

const (
	RetryNum      = 2 // 最大重试次数
	RetryInterval = 1 // 重试时间间隔(秒)
)

const (
	MaxCount     = 10000
	DefaultCount = 1
	DefaultSleep = 1 * time.Second
)

Functions

func Enqueue

func Enqueue(queue string, message interface{}, delayTime time.Duration) (string, error)

Enqueue ...

func GetGoroutineID

func GetGoroutineID() uint64

GetGoroutineID 获取协程ID

func InitClient

func InitClient(opt *redis.Options)

InitClient ...

func InitClusterClient

func InitClusterClient(opt *redis.ClusterOptions)

InitClusterClient ...

func Process

func Process(queue string, delayFunc delayFunc, concurrency int, params ...Params)

Process ...

func RetryEnqueue

func RetryEnqueue(queue string, member interface{}) (string, error)

RetryEnqueue ...

func Run

func Run()

Run ...

func Stats

func Stats(w http.ResponseWriter, req *http.Request)

func StatsServer

func StatsServer(port int)

Types

type EnqueueData

type EnqueueData struct {
	MsgId         string      `json:"msg_id"`
	Message       interface{} `json:"message"`
	RetryNum      uint64      `json:"retry_num"`      // 最大重试次数
	RetryInterval int64       `json:"retry_interval"` // 重试时间间隔(秒)
}

EnqueueData ...

type Params

type Params struct {
	PollSize int64
}

Params ...