go-pkg/asynq
2
0
Fork 0
mirror of https://github.com/hibiken/asynq.git synced 2024-12-26 15:52:18 +08:00
Code Issues Projects Releases Wiki Activity

Implement exponential backoff retry

Browse Source
This commit is contained in:
Ken Hibino 2019-11-17 18:44:40 -08:00
parent 22e2a6f433
commit 6252cc02e0
1 changed files with 16 additions and 5 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

21
asynq.go
View File

@ -15,6 +15,8 @@ import (
"encoding/json" "encoding/json"
"fmt" "fmt"
"log" "log"
"math"
"math/rand"
"strconv" "strconv"
"time" "time"
@ -59,7 +61,7 @@ type taskMessage struct {
// queue name this message should be enqueued to // queue name this message should be enqueued to
Queue string Queue string
// remainig retry count // max number of retry for this task.
Retry int Retry int
// number of times we've retried so far // number of times we've retried so far
@ -151,16 +153,17 @@ func (w *Workers) Run(handler TaskHandler) {
go func(task *Task) { go func(task *Task) {
err := handler(task) err := handler(task)
if err != nil { if err != nil {
if msg.Retry == 0 { if msg.Retried >= msg.Retry {
// TODO(hibiken): Add the task to "dead" collection // TODO(hibiken): Add the task to "dead" collection
fmt.Println("Retry exausted!!!") fmt.Println("Retry exausted!!!")
return return
} }
fmt.Println("RETRY!!!") fmt.Println("RETRY!!!")
delay := 10 * time.Second // TODO(hibiken): Implement exponential backoff. retryAt := time.Now().Add(delaySeconds((msg.Retried)))
msg.Retry-- fmt.Printf("[DEBUG] retying the task in %v\n", retryAt.Sub(time.Now()))
msg.Retried++
msg.ErrorMsg = err.Error() msg.ErrorMsg = err.Error()
if err := zadd(w.rdb, retry, float64(time.Now().Add(delay).Unix()), &msg); err != nil { if err := zadd(w.rdb, retry, float64(retryAt.Unix()), &msg); err != nil {
// TODO(hibiken): Not sure how to handle this error // TODO(hibiken): Not sure how to handle this error
log.Printf("[SEVERE ERROR] could not add msg %+v to 'retry' set: %v\n", msg, err) log.Printf("[SEVERE ERROR] could not add msg %+v to 'retry' set: %v\n", msg, err)
return return
@ -243,3 +246,11 @@ func zadd(rdb *redis.Client, zset string, zscore float64, msg *taskMessage) erro
func listQueues(rdb *redis.Client) []string { func listQueues(rdb *redis.Client) []string {
return rdb.SMembers(allQueues).Val() return rdb.SMembers(allQueues).Val()
} }
// delaySeconds returns a number seconds to delay before retrying.
// Formula taken from https://github.com/mperham/sidekiq.
func delaySeconds(count int) time.Duration {
r := rand.New(rand.NewSource(time.Now().UnixNano()))
s := int(math.Pow(float64(count), 4)) + 15 + (r.Intn(30) * (count + 1))
return time.Duration(s) * time.Second
}