Asynq Distributed Task Queue for Go

To use Asynq, you must define two things: how to create a task and how to process it.

1. Task Creation

A task consists of a type (a string) and a payload (byte slice). It is best practice to wrap task creation in a helper function that marshals your data into JSON.

2. Task Handling

Handlers process the tasks. You can implement them in two ways:

• asynq.HandlerFunc: A simple function with the signature func(context.Context, *asynq.Task) error.
• asynq.Handler interface: A struct that implements the ProcessTask(context.Context, *asynq.Task) error method. This is useful when your handler needs to maintain state or dependencies.

// Task definition example
const TypeEmailDelivery = "email:deliver"

type EmailDeliveryPayload struct {
    UserID     int
    TemplateID string
}

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
}

// HandlerFunc example
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)
    }
    // ... logic
    return nil
}

// Handler interface example
type ImageProcessor struct {}

func (processor *ImageProcessor) ProcessTask(ctx context.Context, t *asynq.Task) error {
    // ... logic
    return nil
}

The PeriodicTaskManager is used to schedule and run tasks at regular intervals by synchronizing with a PeriodicTaskConfigProvider. It manages a background goroutine that periodically calls the provider to fetch the latest task configurations and updates the underlying Scheduler by adding new tasks or removing those no longer present in the configuration.

To use it, you must provide a PeriodicTaskConfigProvider implementation and a Redis connection (either via RedisConnOpt or a RedisUniversalClient).

// Example setup (conceptual)
manager, err := asynq.NewPeriodicTaskManager(asynq.PeriodicTaskManagerOpts{
    PeriodicTaskConfigProvider: myConfigProvider, // Implements PeriodicTaskConfigProvider
    RedisConnOpt:               asynq.RedisConnOpt{Addr: "localhost:6379"},
    SyncInterval:               1 * time.Minute,
})

if err != nil {
    log.Fatal(err)
}

// Run will start the manager and block until an OS signal is received
if err := manager.Run(); err != nil {
    log.Fatal(err)
}

A ServeMux is a multiplexer used to route asynchronous tasks to their appropriate handlers based on the task's type name. It supports pattern matching where longer, more specific patterns take precedence over shorter ones.

Pattern Matching Rules

• Exact Match: If a task type matches a registered pattern exactly, that handler is used.
• Prefix Match: If no exact match is found, the ServeMux searches for the longest pattern that is a prefix of the task type.
• Precedence: Longer patterns win. For example, if you have handlers for images and images:thumbnails, a task of type images:thumbnails:small will be routed to the images:thumbnails handler.
• Fallback: If no pattern matches, the ServeMux returns a NotFoundHandler, which results in an ErrHandlerNotFound error.

// Example of pattern precedence
mux := asynq.NewServeMux()

mux.Handle("images", imagesHandler)
// This handler will be chosen for "images:thumbnails" because it is more specific
mux.Handle("images:thumbnails", thumbnailHandler)

// Task type "images:thumbnails:large" matches "images:thumbnails"
// Task type "images:archive" matches "images"