support NackRedeliveryDelay

Author: HDT3213

README.md

@@ -192,6 +192,15 @@ use WithRetryCount during DelayQueue.SendScheduleMsg or DelayQueue.SendDelayMsg
 queue.SendDelayMsg(msg, time.Hour, delayqueue.WithRetryCount(3))
 ```
 
+### Nack Redelivery Delay
+
+```go
+WithNackRedeliveryDelay(d time.Duration) *DelayQueue
+```
+
+WithNackRedeliveryDelay customizes the interval between redelivery and nack (callback returns false) 
+But if consumption exceeded deadline, the message will be redelivered immediately.
+
 ### Script Preload
 
 ```go

README_CN.md

@@ -183,6 +183,15 @@ WithDefaultRetryCount(count uint)
 queue.SendDelayMsg(msg, time.Hour, delayqueue.WithRetryCount(3))
 ```
 
+### 设置 nack 后重试间隔
+
+```go
+WithNackRedeliveryDelay(d time.Duration) *DelayQueue
+```
+
+WithNackRedeliveryDelay 可以设置 nack (callback 函数返回 false) 之后到重新投递的间隔。
+但是如果消费超时，消息会被立即重新投递。
+
 ### 预加载脚本
 
 ```go

delayqueue.go

@@ -43,7 +43,8 @@ type DelayQueue struct {
 	// for batch consume
 	consumeBuffer chan string
 
-	eventListener EventListener
+	eventListener       EventListener
+	nackRedeliveryDelay time.Duration
 }
 
 // NilErr represents redis nil
@@ -205,6 +206,13 @@ func (q *DelayQueue) WithDefaultRetryCount(count uint) *DelayQueue {
 	return q
 }
 
+// WithNackRedeliveryDelay customizes the interval between redelivery and nack (callback returns false) 
+// If consumption exceeded deadline, the message will be redelivered immediately
+func (q *DelayQueue) WithNackRedeliveryDelay(d time.Duration) *DelayQueue {
+	q.nackRedeliveryDelay = d
+	return q
+}
+
 func (q *DelayQueue) genMsgKey(idStr string) string {
 	if q.useHashTag {
 		return "{dp:" + q.name + "}" + ":msg:" + idStr
@@ -427,7 +435,7 @@ func (q *DelayQueue) nack(idStr string) error {
 	atomic.AddInt32(&q.fetchCount, -1)
 	// update retry time as now, unack2Retry will move it to retry immediately
 	err := q.redisCli.ZAdd(q.unAckKey, map[string]float64{
-		idStr: float64(time.Now().Unix()),
+		idStr: float64(time.Now().Add(q.nackRedeliveryDelay).Unix()),
 	})
 	if err != nil {
 		return fmt.Errorf("negative ack failed: %v", err)

delayqueue_test.go

@@ -30,7 +30,8 @@ func TestDelayQueue_consume(t *testing.T) {
 		WithFetchInterval(time.Millisecond * 50).
 		WithMaxConsumeDuration(0).
 		WithLogger(log.New(os.Stderr, "[DelayQueue]", log.LstdFlags)).
-		WithFetchLimit(2)
+		WithFetchLimit(2).
+		WithNackRedeliveryDelay(time.Second)
 
 	for i := 0; i < size; i++ {
 		err := queue.SendDelayMsg(strconv.Itoa(i), 0, WithRetryCount(retryCount), WithMsgTTL(time.Hour))