delayqueue

README

DelayQueue

DelayQueue is a message queue supporting delayed/scheduled delivery based on redis.

DelayQueue guarantees to deliver at least once.

DelayQueue support ACK/Retry mechanism, it will re-deliver message after a while as long as no confirmation is received. As long as Redis doesn't crash, consumer crashes won't cause message loss.

Example

package main

import (
	"github.com/go-redis/redis/v8"
	"github.com/hdt3213/delayqueue"
	"strconv"
	"time"
)

func main() {
	redisCli := redis.NewClient(&redis.Options{
		Addr: "127.0.0.1:6379",
	})
	queue := delayqueue.NewQueue("example", redisCli, func(payload string) bool {
		// callback returns true to confirm successful consumption.
		// If callback returns false or not return within maxConsumeDuration, DelayQueue will re-deliver this message
		return true
	})
	// send delay message
	for i := 0; i < 10; i++ {
		err := queue.SendDelayMsg(strconv.Itoa(i), time.Hour, delayqueue.WithRetryCount(3))
		if err != nil {
			panic(err)
		}
	}
	// send schedule message
	for i := 0; i < 10; i++ {
		err := queue.SendScheduleMsg(strconv.Itoa(i), time.Now().Add(time.Hour))
		if err != nil {
			panic(err)
		}
	}
	// start consume
	done := queue.StartConsume()
	<-done
}

options

WithLogger(logger *log.Logger)

WithLogger customizes logger for queue

WithFetchInterval(d time.Duration)

WithFetchInterval customizes the interval at which consumer fetch message from redis

WithMaxConsumeDuration(d time.Duration)

WithMaxConsumeDuration customizes max consume duration

If no acknowledge received within WithMaxConsumeDuration after message delivery, DelayQueue will try to deliver this message again

WithFetchLimit(limit uint)

WithFetchLimit limits the max number of messages at one time

Documentation

Index

• func WithRetryCount(count int) interface{}
• type DelayQueue
  • func NewQueue(name string, cli *redis.Client, callback func(string) bool) *DelayQueue
  • func (q *DelayQueue) SendDelayMsg(payload string, duration time.Duration, opts ...interface{}) error
  • func (q *DelayQueue) SendScheduleMsg(payload string, t time.Time, opts ...interface{}) error
  • func (q *DelayQueue) StartConsume() (done <-chan struct{})
  • func (q *DelayQueue) StopConsume()
  • func (q *DelayQueue) WithFetchInterval(d time.Duration) *DelayQueue
  • func (q *DelayQueue) WithFetchLimit(limit uint) *DelayQueue
  • func (q *DelayQueue) WithLogger(logger *log.Logger) *DelayQueue
  • func (q *DelayQueue) WithMaxConsumeDuration(d time.Duration) *DelayQueue

Functions

func WithRetryCount

func WithRetryCount(count int) interface{}

WithRetryCount set retry count for a msg example: queue.SendDelayMsg(payload, duration, delayqueue.WithRetryCount(3))

Types

type DelayQueue

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

DelayQueue is a message queue supporting delayed/scheduled delivery based on redis

func NewQueue

func NewQueue(name string, cli *redis.Client, callback func(string) bool) *DelayQueue

NewQueue creates a new queue, use DelayQueue.StartConsume to consume or DelayQueue.SendScheduleMsg to publish message callback returns true to confirm successful consumption. If callback returns false or not return within maxConsumeDuration, DelayQueue will re-deliver this message

func (*DelayQueue) SendDelayMsg

func (q *DelayQueue) SendDelayMsg(payload string, duration time.Duration, opts ...interface{}) error

SendDelayMsg submits a message delivered after given duration

func (*DelayQueue) SendScheduleMsg

func (q *DelayQueue) SendScheduleMsg(payload string, t time.Time, opts ...interface{}) error

SendScheduleMsg submits a message delivered at given time

func (*DelayQueue) StartConsume

func (q *DelayQueue) StartConsume() (done <-chan struct{})

StartConsume creates a goroutine to consume message from DelayQueue use `<-done` to wait consumer stopping

func (*DelayQueue) StopConsume

func (q *DelayQueue) StopConsume()

StopConsume stops consumer goroutine

func (*DelayQueue) WithFetchInterval

func (q *DelayQueue) WithFetchInterval(d time.Duration) *DelayQueue

WithFetchInterval customizes the interval at which consumer fetch message from redis

func (*DelayQueue) WithFetchLimit

func (q *DelayQueue) WithFetchLimit(limit uint) *DelayQueue

WithFetchLimit limits the max number of messages at one time

func (*DelayQueue) WithLogger

func (q *DelayQueue) WithLogger(logger *log.Logger) *DelayQueue

WithLogger customizes logger for queue

func (*DelayQueue) WithMaxConsumeDuration

func (q *DelayQueue) WithMaxConsumeDuration(d time.Duration) *DelayQueue

WithMaxConsumeDuration customizes max consume duration If no acknowledge received within WithMaxConsumeDuration after message delivery, DelayQueue will try to deliver this message again