Machinery Documentation

Delaying Tasks

Set the ETA field in the Signature to a *time.Time to delay execution.

eta := time.Now().UTC().Add(time.Second * 5)
signature.ETA = &eta

Retrying Tasks

Set RetryCount in the Signature to specify how many times a failed task should be retried. Machinery uses a Fibonacci sequence to space out retries.

signature.RetryCount = 3

Alternatively, return tasks.ErrRetryTaskLater from within the task function to specify a custom delay.

To use Amazon DynamoDB as a result backend in Machinery, you must first create two specific tables with the following primary key configurations:

1. group_metas: Stores metadata for group tasks. The primary key must be set to GroupUUID.
2. task_states: Stores the state of individual tasks. The primary key must be set to TaskUUID.

To install the recommended v2 release, use the following command:

go get github.com/RichardKnop/machinery/v2

If you need to use the legacy v1 version, use:

go get github.com/RichardKnop/machinery

Groups

Execute a set of tasks in parallel. Use tasks.NewGroup and server.SendGroup.

Chords

Execute a group of tasks in parallel, then execute a callback task once the group is finished. Use tasks.NewChord and server.SendChord.

Chains

Execute a sequence of tasks one by one. Each successful task passes its result as arguments to the next task in the chain. Use tasks.NewChain and server.SendChain.

To enable DynamoDB as your result backend, update your Machinery configuration file. You must set result_backend to your DynamoDB endpoint and provide a dynamodb configuration block specifying the names of the tables you created.

Required configuration keys under the dynamodb block:

• task_states_table: The name of the table used for task states.
• group_metas_table: The name of the table used for group task metadata.

broker: 'https://sqs.us-west-1.amazonaws.com/123456789012'
default_queue: machinery-queue
result_backend: 'https://dynamodb.us-west-1.amazonaws.com/123456789012'
results_expire_in: 3600
dynamodb:
  task_states_table: 'task_states'
  group_metas_table: 'group_metas'

Use an AWS SQS URL in the format https://sqs.us-east-2.amazonaws.com/123456789012. The AWS_REGION environment variable must be configured. You can also provide a manually configured SQS Client via SQSConfig.

var sqsClient = sqs.New(session.Must(session.NewSession(&aws.Config{
  Region:         aws.String("YOUR_AWS_REGION"),
  Credentials:    credentials.NewStaticCredentials("YOUR_AWS_ACCESS_KEY", "YOUR_AWS_ACCESS_SECRET", ""),
  HTTPClient:     &http.Client{
    Timeout: time.Second * 120,
  },
})))
var visibilityTimeout = 20
var cnf = &config.Config{
  Broker:          "YOUR_SQS_URL",
  DefaultQueue:    "machinery_tasks",
  ResultBackend:   "YOUR_BACKEND_URL",
  SQS: &config.SQSConfig{
    Client: sqsClient,
    VisibilityTimeout: &visibilityTimeout,
    WaitTimeSeconds: 30,
  },
}

To use AMQP (RabbitMQ), provide a URL in the format amqp://[username:password@]@host[:port].

Available configuration options include:

• Exchange: exchange name (e.g., machinery_exchange)
• ExchangeType: exchange type (e.g., direct)
• QueueBindingArguments: map of additional arguments for binding
• BindingKey: key used to bind the queue to the exchange
• PrefetchCount: number of tasks to prefetch (set to 1 for long-running tasks)
• DelayedQueue: name of the delayed queue for retries or delayed tasks

Use a GCP Pub/Sub URL in the format gcppubsub://YOUR_GCP_PROJECT_ID/YOUR_PUBSUB_SUBSCRIPTION_NAME. You can provide a manually configured Pub/Sub Client via GCPPubSubConfig.

pubsubClient, err := pubsub.NewClient(
    context.Background(),
    "YOUR_GCP_PROJECT_ID",
    option.WithServiceAccountFile("YOUR_GCP_SERVICE_ACCOUNT_FILE"),
)

cnf := &config.Config{
  Broker:          "gcppubsub://YOUR_GCP_PROJECT_ID/YOUR_PUBSUB_SUBSCRIPTION_NAME",
  DefaultQueue:    "YOUR_PUBSUB_TOPIC_NAME",
  ResultBackend:   "YOUR_BACKEND_URL",
  GCPPubSub: config.GCPPubSubConfig{
    Client: pubsubClient,
  },
}

When using DynamoDB as a result backend, you must specify the table names for task states and group metadata. Ensure these tables exist in AWS with TaskUUID and GroupUUID as primary keys respectively.

• TaskStatesTable: Default is task_states.
• GroupMetasTable: Default is group_metas.

Example YAML configuration:

dynamodb:
  task_states_table: 'task_states'
  group_metas_table: 'group_metas'

To start using Machinery, create a Server instance using NewServer. This function automatically initializes the broker, backend, and lock based on the provided config.Config.

Alternatively, if you have already instantiated your own broker, backend, and lock, you can use NewServerWithBrokerBackendLock to inject them directly.

In v2, instead of using a factory, you must inject broker, backend, and lock objects into the machinery.NewServer constructor. This approach avoids importing dependencies for brokers and backends you are not using.

import (
  "github.com/RichardKnop/machinery/v2"
  backendsiface "github.com/RichardKnop/machinery/v2/backends/iface"
  brokersiface "github.com/RichardKnop/machinery/v2/brokers/iface"
  locksiface "github.com/RichardKnop/machinery/v2/locks/iface"
)

var broker brokersiface.Broker
var backend backendsiface.Backend
var lock locksiface.Lock
server := machinery.NewServer(cnf, broker, backend, lock)