mirror of
https://github.com/hibiken/asynq.git
synced 2024-12-24 23:02:18 +08:00
golang基于redis的异步队列
.github/ISSUE_TEMPLATE | ||
internal | ||
tools/asynqmon | ||
.gitignore | ||
.travis.yml | ||
asynq_test.go | ||
asynq.go | ||
background_test.go | ||
background.go | ||
client_test.go | ||
client.go | ||
doc.go | ||
go.mod | ||
go.sum | ||
LICENSE | ||
payload_test.go | ||
payload.go | ||
processor_test.go | ||
processor.go | ||
README.md | ||
scheduler_test.go | ||
scheduler.go |
Asynq
Simple, efficent asynchronous task processing library in Go.
Table of Contents
Overview
Asynq provides a simple interface to asynchronous task processing.
Asynq also ships with a CLI to monitor the queues and take manual actions if needed.
Asynq provides:
- Clear separation of task producer and consumer
- Ability to schedule task processing in the future
- Automatic retry of failed tasks with exponential backoff
- Ability to configure max retry count per task
- Ability to configure max number of worker goroutines to process tasks
- Unix signal handling to safely shutdown background processing
- Enhanced reliability TODO(hibiken): link to wiki page describing this.
- CLI to query and mutate queues state for mointoring and administrative purposes
Requirements
- Redis: 2.6+
- Go: 1.12+
Installation
go get github.com/hibiken/asynq
Getting Started
- Import
asynq
in your file.
import "github.com/hibiken/asynq"
- Create a
Client
instance to create tasks.
func main() {
r := redis.NewClient(&redis.Options{
Addr: "localhost:6379",
}
client := asynq.NewClient(r)
t1 := asynq.Task{
Type: "send_welcome_email",
Payload: map[string]interface{}{
"recipient_id": 1234,
},
}
t2 := asynq.Task{
Type: "send_reminder_email",
Payload: map[string]interface{}{
"recipient_id": 1234,
},
}
// process the task immediately.
err := client.Process(&t1, time.Now())
// process the task 24 hours later.
err = client.Process(&t2, time.Now().Add(24 * time.Hour))
// specify the max number of retry (default: 25)
err = client.Process(&t1, time.Now(), asynq.MaxRetry(1))
}
- Create a
Background
instance to process tasks.
func main() {
bg := asynq.NewBackground(10, &asynq.RedisOpt{
Addr: "localhost:6379",
})
// Blocks until signal TERM or INT is received.
// For graceful shutdown, send signal TSTP to stop processing more tasks
// before sending TERM or INT signal.
bg.Run(handler)
}
The argument to (*asynq.Background).Run
is an interface asynq.Handler
which has one method ProcessTask
.
// ProcessTask should return nil if the processing of a task
// is successful.
//
// If ProcessTask return a non-nil error or panics, the task
// will be retried.
type Handler interface {
ProcessTask(*Task) error
}
The simplest way to implement a handler is to define a function with the same signature and use asynq.HandlerFunc
adapter type when passing it to Run
.
func handler(t *asynq.Task) error {
switch t.Type {
case "send_welcome_email":
id, err := t.Payload.GetInt("recipient_id")
if err != nil {
return err
}
fmt.Printf("Send Welcome Email to %d\n", id)
// ... handle other types ...
default:
return fmt.Errorf("unexpected task type: %s", t.Type)
}
return nil
}
func main() {
bg := asynq.NewBackground(10, &asynq.RedisOpt{
Addr: "localhost:6379",
})
// Use asynq.HandlerFunc adapter for a handler function
bg.Run(asynq.HandlerFunc(handler))
}
License
Asynq is released under the MIT license. See LICENSE.