Getting Started

Welcome to the Tour of Asynq!

In this tutorial, we are going to write two programs, client and workers.

• client.go will create and schedule tasks to be processed asynchronously by the background workers.
• workers.go will start multiple concurrent workers which process the tasks created by the client.

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

Let's start off by creating our two main files.

mkdir quickstart && cd quickstart
go mod init asynq-quickstart
mkdir client workers
touch client/client.go workers/workers.go

And import asynq package.

go get -u github.com/hibiken/asynq

Before we start writing code, let's review a few core types that we'll use in both of our programs.

Redis Connection Option

Asynq uses Redis as a message broker.
Both client.go and workers.go needs to connect to Redis to write and read from it. We are going to use RedisClientOpt specify how to connect to the local Redis instance.

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,
}

Tasks

In asynq, a unit of work is encapsulated in a type 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
}

Type is a simple string value that indicates the type of the given task.
Payload holds data needed for task execution, and you can think of it as map[string]interface{}. One important thing to note is that the payload values have to be serializable.

Now that we've taken a look at the core types, let's start writing our programs.

Client Program

In client.go, we are going to create a few tasks and enqueue them using asynq.Client.

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

asynq.Client supports three methods for scheduling tasks: Enqueue, EnqueueIn, and EnqueueAt.
Use client.Enqueue to enqueue tasks to be processed immediately.
Use client.EnqueueIn or client.EnqueueAt to schedule tasks to be processed in the future.

// client.go
func main() {
    r := &asynq.RedisClientOpt{
        Addr: "localhost:6379",
    }
    client := asynq.NewClient(r)

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

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

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

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

That's all we need for the client program :)

Workers Program

In workers.go, we'll create a asynq.Background instance to start the workers.

NewBackground function takes RedisConnOpt and Config.

Config has quite a few fields to configure how the background task processing should behave.
You can take a look at the documentation on Config to see all the available config options.

To keep it simple, we are only going to specify the concurrency in this example.

// workers.go
func main() {
    r := &asynq.RedisClientOpt{
        Addr: "localhost:6379",
    }
    bg := asynq.NewBackground(r, &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(context.Context, *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(ctx context.Context, t *asynq.Task) error {
    switch t.Type {
    case "email:signup":
        id, err := t.Payload.GetInt("user_id")
        if err != nil {
            return err
        }
        fmt.Printf("Send Welcome Email to User %d\n", id)

    case "email:reminder":
        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() {
    r := &asynq.RedisClientOpt{
        Addr: "localhost:6379",
    }
    bg := asynq.NewBackground(r, &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 use ServeMux to create our handler. Just like the ServeMux from "net/http" package, you register a handler by calling Handle or HandleFunc. ServeMux satisfies the Handler interface, so that you can pass it to (*Background).Run.

// workers.go
func main() {
    mux := asynq.NewServeMux()
    mux.HandleFunc("email:signup", sendWelcomeEmail)
    mux.HandleFunc("email:reminder", sendReminderEmail)

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

func sendWelcomeEmail(context.Context, 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(context.Context, t *asynq.Task) error {
    id, err := t.Payload.GetInt("user_id")
    if err != nil {
        return err
    }
    fmt.Printf("Send Reminder Email to User %d\n", id)
    return nil
}

And that's it for the workers program.

Running the Programs

Now that we have both client and workers, we can run both programs. Let's run the client program to create and schedule tasks.

go run client/client.go

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

Let's use asynqmon tool to inspect the tasks.

asynqmon stats

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

Note: To learn more about the meaning of each state, check out Life of a Task.

Let's run asynqmon with the 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 workers program to process tasks.

go run workers/workers.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 workers.

You should be able to see some text printed in your terminal indicating that the task was processed successfully.

You can run the client again to see how workers pick up the tasks and process them.

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.

Thanks for reading!