Asynq

Simple and efficent asynchronous task processing library in Go.

Important Note: Current major version is zero (v0.x.x) to accomodate rapid development and fast iteration while getting early feedback from users. The public API could change without a major version update before the release of verson 1.0.0.

Table of Contents

• Overview
• Requirements
• Installation
• Getting Started
• Monitoring CLI
• Acknowledgements
• License

Overview

Asynq provides a simple interface to asynchronous task processing.

It also ships with a tool 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
• Automatic failover using Redis sentinels
• Ability to configure max retry count per task
• Ability to configure max number of worker goroutines to process tasks
• Support for priority queues
• Unix signal handling to gracefully shutdown background processing
• CLI tool to query and mutate queues state for mointoring and administrative purposes

Requirements

Dependency	Version
Redis	v2.8+
Go	v1.12+

Installation

To install both asynq library and asynqmon CLI tool, run the following command:

go get -u github.com/hibiken/asynq
go get -u github.com/hibiken/asynq/tools/asynqmon

Getting Started

In this quick tour of asynq, we are going to create two programs.

• producer.go will create and schedule tasks to be processed asynchronously by the consumer.
• consumer.go will process the tasks created by the producer.

This guide assumes that you are running a Redis server at localhost:6379. Before we start, make sure you have Redis installed and running.

1. Import asynq in both files.

import "github.com/hibiken/asynq"

2. Asynq uses Redis as a message broker. Use one of RedisConnOpt types to specify how to connect to Redis. We are going to use RedisClientOpt here.

// both in producer.go and consumer.go
var redis = &asynq.RedisClientOpt{
    Addr: "localhost:6379",
    // Omit if no password is required
    Password: "mypassword",
    // Use a dedicated db number for asynq.
    // By default, Redis offers 16 databases (0..15)
    DB: 0,
}

3. In producer.go, create a Client instance to create and schedule tasks.

// producer.go
func main() {
    client := asynq.NewClient(redis)

    // Create a task with typename and payload.
    t1 := asynq.NewTask(
        "send_welcome_email",
        map[string]interface{}{"user_id": 42})

    t2 := asynq.NewTask(
        "send_reminder_email",
        map[string]interface{}{"user_id": 42})

    // Process the task immediately.
    err := client.Schedule(t1, time.Now())
    if err != nil {
        log.Fatal(err)
    }

    // Process the task 24 hours later.
    err = client.Schedule(t2, time.Now().Add(24 * time.Hour))
    if err != nil {
        log.Fatal(err)
    }
}

4. In consumer.go, create a Background instance to process tasks.

// consumer.go
func main() {
    bg := asynq.NewBackground(redis, &asynq.Config{
        Concurrency: 10,
    })

    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("user_id")
        if err != nil {
            return err
        }
        fmt.Printf("Send Welcome Email to User %d\n", id)

    case "send_reminder_email":
        id, err := t.Payload.GetInt("user_id")
        if err != nil {
            return err
        }
        fmt.Printf("Send Reminder Email to User %d\n", id)

    default:
        return fmt.Errorf("unexpected task type: %s", t.Type)
    }
    return nil
}

func main() {
    bg := asynq.NewBackground(redis, &asynq.Config{
        Concurrency: 10,
    })

    // Use asynq.HandlerFunc adapter for a handler function
    bg.Run(asynq.HandlerFunc(handler))
}

We could kep adding cases to this handler function, but in a realistic application, it's convenient to define the logic for each case in a separate function.

To refactor our code, let's create a simple dispatcher which maps task type to its handler.

// consumer.go

// Dispatcher is used to dispatch tasks to registered handlers.
type Dispatcher struct {
    mapping map[string]asynq.HandlerFunc
}

// HandleFunc registers a task handler
func (d *Dispatcher) HandleFunc(taskType string, fn asynq.HandlerFunc) {
    d.mapping[taskType] = fn
}

// ProcessTask processes a task.
//
// NOTE: Dispatcher satisfies asynq.Handler interface.
func (d *Dispatcher) ProcessTask(task *asynq.Task) error {
    fn, ok := d.mapping[task.Type]
    if !ok {
        return fmt.Errorf("no handler registered for %q", task.Type)
    }
    return fn(task)
}

func main() {
    d := &Dispatcher{mapping: make(map[string]asynq.HandlerFunc)}
    d.HandleFunc("send_welcome_email", sendWelcomeEmail)
    d.HandleFunc("send_reminder_email", sendReminderEmail)

    bg := asynq.NewBackground(redis, &asynq.Config{
        Concurrency: 10,
    })
    bg.Run(d)
}

func sendWelcomeEmail(t *asynq.Task) error {
    id, err := t.Payload.GetInt("user_id")
    if err != nil {
        return err
    }
    fmt.Printf("Send Welcome Email to User %d\n", id)
    return nil
}

func sendReminderEmail(t *asynq.Task) error {
    id, err := t.Payload.GetInt("user_id")
    if err != nil {
        return err
    }
    fmt.Printf("Send Welcome Email to User %d\n", id)
    return nil
}

Now that we have both task producer and consumer, we can run both programs.

go run consumer.go

Note: This will not exit until you send a signal to terminate the program. See Signal Wiki page for best practice on how to safely terminate background processing.

With our consumer running, also run

go run producer.go

This will create a task and the first task will get processed immediately by the consumer. The second task will be processed 24 hours later.

Let's use asynqmon tool to inspect the tasks.

asynqmon stats

This command will show the number of tasks in each state and stats for the current date as well as redis information.

To understand the meaning of each state, see Life of a Task Wiki page.

For in-depth guide on asynqmon tool, see the README for the CLI.

This was a quick tour of asynq basics. To see all of its features such as priority queues and custom retry, see the Wiki page.

Monitoring CLI

Asynq ships with a CLI tool to inspect the state of queues and tasks.

To install the CLI, run the following command:

go get github.com/hibiken/asynq/tools/asynqmon

For details on how to use the tool, see the README for the asynqmon CLI.

Acknowledgements

• Sidekiq : Many of the design ideas are taken from sidekiq and its Web UI
• Cobra : Asynqmon CLI is built with cobra

License

Asynq is released under the MIT license. See LICENSE.