asynq README

The project's README from github.com/hibiken/asynq. As one markdown file: README.md.

Asynq logo

Simple, reliable & efficient distributed task queue in Go

GoDoc Go Report Card Build Status License: MIT Gitter chat

Asynq is a Go library for queueing tasks and processing them asynchronously with workers. It's backed by Redis and is designed to be scalable yet easy to get started.

Highlevel overview of how Asynq works:

Task queues are used as a mechanism to distribute work across multiple machines. A system can consist of multiple worker servers and brokers, giving way to high availability and horizontal scaling.

Example use case

Task Queue Diagram

Features

Stability and Compatibility

Status: The library relatively stable and is currently undergoing moderate development with less frequent breaking API changes.

☝️ Important Note: Current major version is zero (v0.x.x) to accommodate rapid development and fast iteration while getting early feedback from users (feedback on APIs are appreciated!). The public API could change without a major version update before v1.0.0 release.

Redis Cluster Compatibility

Some of the lua scripts in this library may not be compatible with Redis Cluster.

Sponsoring

If you are using this package in production, please consider sponsoring the project to show your support!

Quickstart

Make sure you have Go installed (download). The last two Go versions are supported (See https://go.dev/dl).

Initialize your project by creating a folder and then running go mod init github.com/your/repo (learn more) inside the folder. Then install Asynq library with the go get command:

go get -u github.com/hibiken/asynq

Make sure you're running a Redis server locally or from a Docker container. Version 4.0 or higher is required.

Next, write a package that encapsulates task creation and task handling.

package tasks

import (
    "context"
    "encoding/json"
    "fmt"
    "log"
    "time"
    "github.com/hibiken/asynq"
)

// A list of task types.
const (
    TypeEmailDelivery   = "email:deliver"
    TypeImageResize     = "image:resize"
)

type EmailDeliveryPayload struct {
    UserID     int
    TemplateID string
}

type ImageResizePayload struct {
    SourceURL string
}

//----------------------------------------------
// Write a function NewXXXTask to create a task.
// A task consists of a type and a payload.
//----------------------------------------------

func NewEmailDeliveryTask(userID int, tmplID string) (*asynq.Task, error) {
    payload, err := json.Marshal(EmailDeliveryPayload{UserID: userID, TemplateID: tmplID})
    if err != nil {
        return nil, err
    }
    return asynq.NewTask(TypeEmailDelivery, payload), nil
}

func NewImageResizeTask(src string) (*asynq.Task, error) {
    payload, err := json.Marshal(ImageResizePayload{SourceURL: src})
    if err != nil {
        return nil, err
    }
    // task options can be passed to NewTask, which can be overridden at enqueue time.
    return asynq.NewTask(TypeImageResize, payload, asynq.MaxRetry(5), asynq.Timeout(20 * time.Minute)), nil
}

//---------------------------------------------------------------
// Write a function HandleXXXTask to handle the input task.
// Note that it satisfies the asynq.HandlerFunc interface.
//
// Handler doesn't need to be a function. You can define a type
// that satisfies asynq.Handler interface. See examples below.
//---------------------------------------------------------------

func HandleEmailDeliveryTask(ctx context.Context, t *asynq.Task) error {
    var p EmailDeliveryPayload
    if err := json.Unmarshal(t.Payload(), &p); err != nil {
        return fmt.Errorf("json.Unmarshal failed: %v: %w", err, asynq.SkipRetry)
    }
    log.Printf("Sending Email to User: user_id=%d, template_id=%s", p.UserID, p.TemplateID)
    // Email delivery code ...
    return nil
}

// ImageProcessor implements asynq.Handler interface.
type ImageProcessor struct {
    // ... fields for struct
}

func (processor *ImageProcessor) ProcessTask(ctx context.Context, t *asynq.Task) error {
    var p ImageResizePayload
    if err := json.Unmarshal(t.Payload(), &p); err != nil {
        return fmt.Errorf("json.Unmarshal failed: %v: %w", err, asynq.SkipRetry)
    }
    log.Printf("Resizing image: src=%s", p.SourceURL)
    // Image resizing code ...
    return nil
}

func NewImageProcessor() *ImageProcessor {
	return &ImageProcessor{}
}

In your application code, import the above package and use Client to put tasks on queues.

package main

import (
    "log"
    "time"

    "github.com/hibiken/asynq"
    "your/app/package/tasks"
)

const redisAddr = "127.0.0.1:6379"

func main() {
    client := asynq.NewClient(asynq.RedisClientOpt{Addr: redisAddr})
    defer client.Close()

    // ------------------------------------------------------
    // Example 1: Enqueue task to be processed immediately.
    //            Use (*Client).Enqueue method.
    // ------------------------------------------------------

    task, err := tasks.NewEmailDeliveryTask(42, "some:template:id")
    if err != nil {
        log.Fatalf("could not create task: %v", err)
    }
    info, err := client.Enqueue(task)
    if err != nil {
        log.Fatalf("could not enqueue task: %v", err)
    }
    log.Printf("enqueued task: id=%s queue=%s", info.ID, info.Queue)


    // ------------------------------------------------------------
    // Example 2: Schedule task to be processed in the future.
    //            Use ProcessIn or ProcessAt option.
    // ------------------------------------------------------------

    info, err = client.Enqueue(task, asynq.ProcessIn(24*time.Hour))
    if err != nil {
        log.Fatalf("could not schedule task: %v", err)
    }
    log.Printf("enqueued task: id=%s queue=%s", info.ID, info.Queue)


    // ----------------------------------------------------------------------------
    // Example 3: Set other options to tune task processing behavior.
    //            Options include MaxRetry, Queue, Timeout, Deadline, Unique etc.
    // ----------------------------------------------------------------------------

    task, err = tasks.NewImageResizeTask("https://example.com/myassets/image.jpg")
    if err != nil {
        log.Fatalf("could not create task: %v", err)
    }
    info, err = client.Enqueue(task, asynq.MaxRetry(10), asynq.Timeout(3 * time.Minute))
    if err != nil {
        log.Fatalf("could not enqueue task: %v", err)
    }
    log.Printf("enqueued task: id=%s queue=%s", info.ID, info.Queue)
}

Next, start a worker server to process these tasks in the background. To start the background workers, use Server and provide your Handler to process the tasks.

You can optionally use ServeMux to create a handler, just as you would with net/http Handler.

package main

import (
    "log"

    "github.com/hibiken/asynq"
    "your/app/package/tasks"
)

const redisAddr = "127.0.0.1:6379"

func main() {
    srv := asynq.NewServer(
        asynq.RedisClientOpt{Addr: redisAddr},
        asynq.Config{
            // Specify how many concurrent workers to use
            Concurrency: 10,
            // Optionally specify multiple queues with different priority.
            Queues: map[string]int{
                "critical": 6,
                "default":  3,
                "low":      1,
            },
            // See the godoc for other configuration options
        },
    )

    // mux maps a type to a handler
    mux := asynq.NewServeMux()
    mux.HandleFunc(tasks.TypeEmailDelivery, tasks.HandleEmailDeliveryTask)
    mux.Handle(tasks.TypeImageResize, tasks.NewImageProcessor())
    // ...register other handlers...

    if err := srv.Run(mux); err != nil {
        log.Fatalf("could not run server: %v", err)
    }
}

For a more detailed walk-through of the library, see our Getting Started guide.

To learn more about asynq features and APIs, see the package godoc.

Web UI

Asynqmon is a web based tool for monitoring and administrating Asynq queues and tasks.

Here's a few screenshots of the Web UI:

Queues view

Web UI Queues View

Tasks view

Web UI TasksView

Metrics view Screen Shot 2021-12-19 at 4 37 19 PM

Settings and adaptive dark mode

Web UI Settings and adaptive dark mode

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

Command Line Tool

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

To install the CLI tool, run the following command:

go install github.com/hibiken/asynq/tools/asynq@latest

Here's an example of running the asynq dash command:

Gif

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

Contributing

We are open to, and grateful for, any contributions (GitHub issues/PRs, feedback on Gitter channel, etc) made by the community.

Please see the Contribution Guide before contributing.

License

Copyright (c) 2019-present Ken Hibino and Contributors. Asynq is free and open-source software licensed under the MIT License. Official logo was created by Vic Shóstak and distributed under Creative Commons license (CC0 1.0 Universal).

asynq (hibiken/asynq)

Searching Database…

13.8K ·GitHub
Asynq
Что это такое?Asynq — это Go-библиотека для постановки задач в очередь и их асинхронной обработки воркерами, с Redis в качестве хранилища. Она распределяет работу между несколькими машинами и рассчитана на масштабируемость, но при этом на неё легко начать.
Это библиотека, которую я импортирую в своё Go-приложение, или отдельный сервис, который нужно разворачивать?Это Go-библиотека: ваш код ставит задачи в очередь через Client, а вы запускаете worker-серверы, которые забирают задачи и запускают по горутине на каждую. Несколько worker-серверов и брокеров дают отказоустойчивость и горизонтальное масштабирование.
Как это запустить?Установите Go (поддерживаются две последние версии), инициализируйте проект командой go mod init и добавьте Asynq через go get. Нужен Redis-сервер версии 4.0 или выше — локально или в Docker; дальше ставьте задачи в очередь через Client и обрабатывайте их через Server.
Есть ли подводные камни, прежде чем я начну на этом что-то строить?Публичный API пока в версии v0.x и до v1.0 может измениться без повышения мажорной версии. Кроме того, некоторые Lua-скрипты могут быть несовместимы с Redis Cluster, поэтому проверьте свою конфигурацию Redis.
Как следить за очередями и задачами, когда всё уже запущено?Asynqmon — это веб-интерфейс для мониторинга и управления очередями и задачами Asynq, а вместе с библиотекой идёт CLI, чтобы просматривать их и управлять ими удалённо. Есть также интеграция с Prometheus для сбора и визуализации метрик очередей.
Что произойдёт, если задача упадёт с ошибкой или воркер сломается посреди задачи?Каждая задача гарантированно выполняется хотя бы один раз: упавшие задачи повторяются, а при падении воркера задачи автоматически восстанавливаются. Для каждой задачи можно также задать таймаут и дедлайн.
Можно ли запланировать задачи на потом и сделать одни очереди приоритетнее других?Да. Asynq поддерживает отложенные и периодические задачи, взвешенные и строгие приоритеты очередей, а также приостановку очереди, чтобы остановить обработку задач из неё.
Можно ли защититься от дубликатов задач в очереди?Да, опция unique не пропускает дубликаты задач. Агрегация также может группировать задачи, чтобы несколько последовательных операций обрабатывались одним пакетом.
What are similar projects?rq, bull, nsq, automq, effect-mq, sidekiq, rabbitmq-server, emqx
Which topics is it in?task-queue, queue, background-jobs, asynchronous, high-availability, scalability, prometheus, redis, golang, distributed-systems, monitoring, cloud
Creator
hibiken
Size
8.2 MB
Stars
13.8K
Updated
CommitOct 19Releasev0.26.0Feb 03
Age
6y
Commits · 90 Days