support NackRedeliveryDelay

This commit is contained in:
finley
2025-01-19 08:08:41 +08:00
parent 04a15cbd8d
commit 828cdbe6ff
4 changed files with 30 additions and 3 deletions

View File

@@ -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

View File

@@ -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

View File

@@ -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)

View File

@@ -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))