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 v1.0.0 release.

Table of Contents

• Overview
• Requirements
• Installation
• Getting Started
• Command Line Tool
• Acknowledgements
• License

Overview

Package asynq provides a framework for asynchronous task processing.

Asynq provides:

• Clear separation of task producer and consumer
• Ability to process multiple tasks concurrently
• Ability to schedule task processing in the future
• Automatic retry of failed tasks with exponential backoff
• Ability to configure task retry count and retry delay
• Support for priority queues
• Unix signal handling to gracefully shutdown background processing
• Automatic failover using Redis sentinels
• Command line tool to query tasks for monitoring and troubleshooting purposes

Requirements

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

Installation

To install both asynq library and asynqmon command line 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, we are going to create a Client instance to create and schedule tasks.

   In asynq, a unit of work to be performed is encapsluated in a message called Task. Which has two fields: Type and Payload.

// Task represents a task to be performed.
type Task struct {
    // Type indicates the type of task to be performed.
    Type string

    // Payload holds data needed to perform the task.
    Payload Payload
}

To create a task, use NewTask function and pass type and payload for the task.

You can schedule a task by calling Schedule on a client passing the task and the time to be performed.

// 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.

NewBackground function takes RedisConnOpt and Config.

You can take a look at documentation on Config to see the available options.

We are only going to specify the concurrency in this example.

// 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 keep 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 producer.go

This will create two tasks: One that should processed immediately and another to be processed 24 hours later.

Let's use asynqmon tool to inspect the tasks.

asynqmon stats

You should able to see that there's one task in Enqueued state and another in Scheduled state.

Note: To understand the meaning of each state, see Life of a Task on our Wiki page.

Let's run asynqmon with watch command so that we can continuously run the command to see the changes.

watch -n 3 asynqmon stats # Runs `asynqmon stats` every 3 seconds

And finally, let's start the consumer program to process scheduled tasks.

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.

This was a whirlwind tour of asynq basics. To learn more about all of its features such as priority queues and custom retry, see our Wiki page.

Command Line Tool

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

To install, run the following command:

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

For details on how to use the tool, refer to the tool's README.

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.