machinery
Asynchronous task queue/job queue based on distributed message passing
About machinery
This large README is shown as a plain-text preview. Read the full README at the source
[1]: https://raw.githubusercontent.com/RichardKnop/assets/master/machinery/example_worker.png
[2]: https://raw.githubusercontent.com/RichardKnop/assets/master/machinery/example_worker_receives_tasks.png
[3]: http://patreon_public_assets.s3.amazonaws.com/sized/becomeAPatronBanner.png
## Machinery
Machinery is an asynchronous task queue/job queue based on distributed message passing.
[](http://godoc.org/github.com/RichardKnop/machinery/v1)
[](https://codecov.io/gh/RichardKnop/machinery)
[](https://goreportcard.com/report/github.com/RichardKnop/machinery)
[](http://opentracing.io)
[](https://sourcegraph.com/github.com/RichardKnop/machinery?badge)
[](https://richardknop.github.io/donate/)
---
* [V2 Experiment](#v2-experiment)
* [First Steps](#first-steps)
* [Configuration](#configuration)
* [Lock](#lock)
* [Broker](#broker)
* [DefaultQueue](#defaultqueue)
* [ResultBackend](#resultbackend)
* [ResultsExpireIn](#resultsexpirein)
* [AMQP](#amqp-2)
* [DynamoDB](#dynamodb)
* [Redis](#redis-2)
* [GCPPubSub](#gcppubsub)
* [Custom Logger](#custom-logger)
* [Server](#server)
* [Workers](#workers)
* [Tasks](#tasks)
* [Registering Tasks](#registering-tasks)
* [Signatures](#signatures)
* [Supported Types](#supported-types)
* [Sending Tasks](#sending-tasks)
* [Delayed Tasks](#delayed-tasks)
* [Retry Tasks](#retry-tasks)
* [Get Pending Tasks](#get-pending-tasks)
* [Keeping Results](#keeping-results)
* [Workflows](#workflows)
* [Groups](#groups)
* [Chords](#chords)
* [Chains](#chains)
* [Periodic Tasks & Workflows](#periodic-tasks--workflows)
* [Periodic Tasks](#periodic-tasks)
* [Periodic Groups](#periodic-groups)
* [Periodic Chains](#periodic-chains)
* [Periodic Chords](#periodic-chords)
* [Development](#development)
* [Requirements](#requirements)
* [Dependencies](#dependencies)
* [Testing](#testing)
### V2
I recommend using V2 in order to avoid having to import all dependencies for brokers and backends you are not using.
Instead of factory, you will need to inject broker and backend objects to the server constructor:
```go
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)
// server.NewWorker("machinery", 10)
```
### First Steps
To install recommended v2 release:
```sh
go get github.com/RichardKnop/machinery/v2
```
If you want to use legacy v1 version, you still can:
```sh
go get github.com/RichardKnop/machinery
```
First, you will need to define some tasks. Look at sample tasks in `v2/example/tasks/tasks.go` to see a few examples.
Second, you will need to launch a worker process with one of these commands (v2 is recommended since it doesn't import dependencies for all brokers / backends, only those you actually need):
```sh
cd v2/
go run example/amqp/main.go worker
go run example/redigo/main.go worker // Redis with redigo driver
go run example/go-redis/main.go worker // Redis with Go Redis driver
go run example/amqp/main.go worker
go run example/redis/main.go worker
```
![Example worker][1]
Finally, once you have a worker running and waiting for tasks to consume, send some tasks with one of these commands (v2 is recommended since it doesn't import dependencies for all brokers / backends, only those you actually need):
```sh
cd v2
go run v2/example/amqp/main.go send
go run v2/example/redigo/main.go send // Redis with redigo driver
go run v2/example/go-redis/main.go send // Redis with Go Redis driver
```
You will be able to see the tasks being processed asynchronously by the worker:
![Example worker receives tasks][2]
### Configuration
The [config](/v2/config/config.go) package has convenience methods for loading configuration from environment variables or a YAML file. For example, load configuration from environment variables:
```go
cnf, err := config.NewFromEnvironment()
```
Or load from YAML file:
```go
cnf, err := config.NewFromYaml("config.yml", true)
```
Second boolean flag enables live reloading of configuration every 10 seconds. Use `false` to disable live reloading.
Machinery configuration is encapsulated by a `Config` struct and injected as a dependency to objects that need it.
#### Lock
##### Redis
Use Redis URL in one of these formats:
```
redis://[password@]host[port][/db_num]
```
For example:
1. `redis://localhost:6379`, or with password `redis://password@localhost:6379`
#### Broker
A message broker. Currently supported brokers are:
##### AMQP
Use AMQP URL in the format:
```
amqp://[username:password@]@host[:port]
```
For example:
1. `amqp://guest:guest@localhost:5672`
AMQP also supports multiples brokers urls. You need to specify the URL separator in the `MultipleBrokerSeparator` field.
##### Redis
Use Redis URL in one of these formats:
```
redis://[password@]host[port][/db_num]
redis+socket://[password@]/path/to/file.sock[:/db_num]
```
For example:
1. `redis://localhost:6379`, or with password `redis://password@localhost:6379`
2. `redis+socket://password@/path/to/file.sock:/0`
##### AWS SQS
Use AWS SQS URL in the format:
```
https://sqs.us-east-2.amazonaws.com/123456789012
```
See [AWS SQS docs](https://docs.aws.amazon.com/sdk-for-go/v1/developer-guide/configuring-sdk.html) for more information.
Also, configuring `AWS_REGION` is required, or an error would be thrown.
To use a manually configured SQS Client:
```go
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,
// if VisibilityTimeout is nil default to the overall visibility timeout setting for the queue
// https://docs.aws.amazon.com/AWSSimpleQueueService/latest/SQSDeveloperGuide/sqs-visibility-timeout.html
VisibilityTimeout: &visibilityTimeout,
WaitTimeSeconds: 30,
},
}
```
##### GCP Pub/Sub
Use GCP Pub/Sub URL in the format:
```
gcppubsub://YOUR_GCP_PROJECT_ID/YOUR_PUBSUB_SUBSCRIPTION_NAME
```
To use a manually configured Pub/Sub Client:
```go
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,
},
}
```
#### DefaultQueue
Default queue name, e.g. `machinery_tasks`.
#### ResultBackend
Result backend to use for keeping task states and results.
Currently supported backends are:
##### Redis
Use Redis URL in one of these formats:
```
redis://[password@]host[port][/db_num]
redis+socket://[password@]/path/to/file.sock[:/db_num]
```
For example:
1. `redis://localhost:6379`, or with password `redis://password@localhost:6379`
2. `redis+socket://password@/path/to/file.sock:/0`
3. cluster/sentinel `redis://host1:port1,host2:port2,host3:port3/0`
4. cluster/sentinel with password `redis://pass@host1:port1,host2:port2,host3:port3/0`
##### Memcache
Use Memcache URL in the format:
```
memcache://host1[:port1][,host2[:port2],...[,hostN[:portN]]]
```
For example:
1. `memcache://localhost:11211` for a single instance, or
2. `memcache://10.0.0.1:11211,10.0.0.2:11211` for a cluster
##### AMQP
Use AMQP URL in the format:
```
amqp://[username:password@]@host[:port]
```
For example:
1. `amqp://guest:guest@localhost:5672`
> Keep in mind AMQP is not recommended as a result backend. See [Keeping Results](https://github.com/RichardKnop/machinery#keeping-results)
##### MongoDB
Use Mongodb URL in the format:
```
mongodb://[username:password@]host1[:port1][,host2[:port2],...[,hostN[:portN]]][/[database][?options]]
```
For example:
1. `mongodb://localhost:27017/taskresults`
See [MongoDB docs](https://docs.mongodb.org/manual/reference/connection-string/) for more information.
#### ResultsExpireIn
How long to store task results for in seconds. Defaults to `3600` (1 hour).
#### AMQP
RabbitMQ related configuration. Not necessary if you are using other broker/backend.
* `Exchange`: exchange name, e.g. `machinery_exchange`
* `ExchangeType`: exchange type, e.g. `direct`
* `QueueBindingArguments`: an optional map of additional arguments used when binding to an AMQP queue
* `BindingKey`: The queue is bind to the exchange with this key, e.g. `machinery_task`
* `PrefetchCount`: How many tasks to prefetch (set to `1` if you have long running tasks)
* `DelayedQueue`: delayed queue name to be used for task retry or delayed task (if empty it will follow auto create and delate delayed queues)
#### DynamoDB
DynamoDB related configuration. Not necessary if you are using other backend.
* `TaskStatesTable`: Custom table name for saving task states. Default one is `task_states`, and make sure to create this table in your AWS admin first, using `TaskUUID` as table's primary key.
* `GroupMetasTable`: Custom table name for saving group metas. Default one is `group_metas`, and make sure to create this table in your AWS admin first, using `GroupUUID` as table's primary key.
For example:
```
dynamodb:
task_states_table: 'task_states'
group_metas_table: 'group_metas'
```
If these tables are not found, an fatal error would be thrown.
If you wish to expire the records, you can configure the `TTL` field in AWS admin for these tables. The `TTL` field is set based on the `ResultsExpireIn` value in the Server's config. See https://docs.aws.amazon.com/amazondynamodb/latest/developerguide/howitworks-ttl.html for more information.
#### Redis
Redis related configuration. Not necessary if you are using other backend.
See: [config](/v1/config/config.go) (TODO)
#### GCPPubSub
GCPPubSub related configuration. Not necessary if you are using other backend.
See: [config](/v1/config/config.go) (TODO)
### Custom Logger
You can define a custom logger by implementing the following interface:
```go
type Interface interface {
Print(...interface{})
Printf(string, ...interface{})
Println(...interface{})
Fatal(...interface{})
Fatalf(string, ...interface{})
Fatalln(...interface{})
Panic(...interface{})
Panicf(string, ...interface{})
Panicln(...interface{})
}
```
Then just set the logger in your setup code by calling `Set` function exported by `github.com/RichardKnop/machinery/v1/log` package:
```go
log.Set(myCustomLogger)
```
### Server
A Machinery library must be instantiated before use. The way this is done is by creating a `Server` instance. `Server` is a base object which stores Machinery configuration and registered tasks. E.g.:
```go
import (
"github.com/RichardKnop/machinery/v1/config"
"github.com/RichardKnop/machinery/v1"
)
var cnf = &config.Config{
Broker: "amqp://guest:guest@localhost:5672/",
DefaultQueue: "machinery_tasks",
ResultBackend: "amqp://guest:guest@localhost:5672/",
AMQP: &config.AMQPConfig{
Exchange: "machinery_exchange",
ExchangeType: "direct",
BindingKey: "machinery_task",
},
}
server, err := machinery.NewServer(cnf)
if err != nil {
// do something with the error
}
```
### Workers
In order to consume tasks, you need to have one or more workers running. All you need to run a worker is a `Server` instance with registered tasks. E.g.:
```go
worker := server.NewWorker("worker_name", 10)
err := worker.Launch()
if err != nil {
// do something with the error
}
```
Each worker will only consume registered tasks. For each task on the queue the Worker.Process() method will be run
in a goroutine. Use the second parameter of `server.NewWorker` to limit the number of concurrently running Worker.Process()
calls (per worker). Example: 1 will serialize task execution while 0 makes the number of concurrently executed tasks unlimited (default).
### Tasks
Tasks are a building block of Machinery applications. A task is a function which defines what happens when a worker receives a message.
Each task needs to return an error as a last return value. In addition to error tasks can now return any number of arguments.
Examples of valid tasks:
```go
func Add(args ...int64) (int64, error) {
sum := int64(0)
for _, arg := range args {
sum += arg
}
return sum, nil
}
func Multiply(args ...int64) (int64, error) {
sum := int64(1)
for _, arg := range args {
sum *= arg
}
return sum, nil
}
// You can use context.Context as first argument to tasks, useful for open tracing
func TaskWithContext(ctx context.Context, arg Arg) error {
// ... use ctx ...
return nil
}
// Tasks need to return at least error as a minimal requirement
func DummyTask(arg string) error {
return errors.New(arg)
}
// You can also return multiple results from the task
func DummyTask2(arg1, arg2 string) (string, string, error) {
return arg1, arg2, nil
}
```
#### Registering Tasks
Before your workers can consume a task, you need to register it with the server. This is done by assigning a task a unique name:
```go
server.RegisterTasks(map[string]interface{}{
"add": Add,
"multiply": Multiply,
})
```
Tasks can also be registered one by one:
```go
server.RegisterTask("add", Add)
server.RegisterTask("multiply", Multiply)
```
Simply put, when a worker receives a message like this:
```json
{
"UUID": "48760a1a-8576-4536-973b-da09048c2ac5",
"Name": "add",
"RoutingKey": "",
"ETA": null,
"GroupUUID": "",
"GroupTaskCount": 0,
"Args": [
{
"Type": "int64",
"Value": 1,
},
{
"Type": "int64",
"Value": 1,
}
],
"Immutable": false,
"RetryCount": 0,
"RetryTimeout": 0,
"OnSuccess": null,
"OnError": null,
"ChordCallback": null
}
```
It will call Add(1, 1). Each task should return an error as well so we can handle failures.
Ideally, tasks should be idempotent which means there will be no unintended consequences when a task is called multiple times with the same arguments.
#### Signatures
A signature wraps calling arguments, execution options (such as immutability) and success/error callbacks of a task so it can be sent across the wire to workers. Task signatures implement a simple interface:
```go
// Arg represents a single argument passed to invocation fo a task
type Arg struct {
Type string
Value interface{}
}
// Headers represents the headers which should be used to direct the task
type Headers map[string]interface{}
// Signature represents a single task invocation
type Signature struct {
UUID string
Name string
RoutingKey string
ETA *time.Time
GroupUUID string
GroupTaskCount int
Args []Arg
Headers Headers
Immutable bool
RetryCount int
RetryTimeout int
OnSuccess []*Signature
OnError []*Signature
ChordCallback *Signature
}
```
`UUID` is a unique ID of a task. You can either set it yourself or it will be automatically generated.
`Name` is the unique task name by which it is registered against a Server instance.
`RoutingKey` is used for routing a task to correct queue. If you leave it empty, the default behaviour will be to set it to the default queue's binding key for direct exchange type and to the default queue name for other exchange types.
`ETA` is a timestamp used for delaying a task. if it's nil, the task will be published for workers to consume immediately. If it is set, the task will be delayed until the ETA timestamp.
`GroupUUID`, `GroupTaskCount` are useful for creating groups of tasks.
`Args` is a list of arguments that will be passed to the task when it is executed by a worker.
`Headers` is a list of headers that will be used when publishing the task to AMQP queue.
`Immutable` is a flag which defines whether a result of the executed task can be modified or not. This is important with `OnSuccess` callbacks. Immutable task will not pass its result to its success callbacks while a mutable task will prepend its result to args sent to callback tasks. Long story short, set Immutable to false if you want to pass result of the first task in a chain to the second task.
`RetryCount` specifies how many times a failed task should be retried (defaults to 0). Retry attempts will be spaced out in time, after each failure another attempt will be scheduled further to the future.
`RetryTimeout` specifies how long to wait before resending task to the queue for retry attempt. Default behaviour is to use fibonacci sequence to increase the timeout after each failed retry attempt.
`OnSuccess` defines tasks which will be called after the task has executed successfully. It is a slice of task signature structs.
`OnError` defines tasks which will be called after the task execution fails. The first argument passed to error callbacks will be the error string returned from the failed task.
`ChordCallback` is used to create a callback to a group of tasks.
#### Supported Types
Machinery encodes tasks to JSON before sending them to the broker. Task results are also stored in the backend as JSON encoded strings. Therefor only types with native JSON representation can be supported. Currently supported types are:
* `bool`
* `int`
* `int8`
* `int16`
* `int32`
* `int64`
* `uint`
* `uint8`
* `uint16`
* `uint32`
* `uint64`
* `float32`
* `float64`
* `string`
* `[]bool`
* `[]int`
* `[]int8`
* `[]int16`
* `[]int32`
* `[]int64`
* `[]uint`
* `[]uint8`
* `[]uint16`
* `[]uint32`
* `[]uint64`
* `[]float32`
* `[]float64`
* `[]string`
#### Sending Tasks
Tasks can be called by passing an instance of `Signature` to an `Server` instance. E.g:
```go
import (
"github.com/RichardKnop/machinery/v1/tasks"
)
signature := &tasks.Signature{
Name: "add",
Args: []tasks.Arg{
{
Type: "int64",
Value: 1,
},
{
Type: "int64",
Value: 1,
},
},
}
asyncResult, err := server.SendTask(signature)
if err != nil {
// failed to send the task
// do something with the error
}
```
#### Delayed Tasks
You can delay a task by setting the `ETA` timestamp field on the task signature.
```go
// Delay the task by 5 seconds
eta := time.Now().UTC().Add(time.Second * 5)
signature.ETA = &eta
```
#### Retry Tasks
You can set a number of retry attempts before declaring task as failed. Fibonacci sequence will be used to space out retry requests over time. (See `RetryTimeout` for details.)
```go
// If the task fails, retry it up to 3 times
signature.RetryCount = 3
```
Alternatively, you can return `tasks.ErrRetryTaskLater` from your task and specify duration after which the task should be retried, e.g.:
```go
return tasks.NewErrRetryTaskLater("some error", 4 * time.Hour)
```
#### Get Pending Tasks
Tasks currently waiting in the queue to be consumed by workers can be inspected, e.g.:
```goFrequently Asked Questions
What is machinery?
machinery is a Messaging library for the Go programming language. Asynchronous task queue/job queue based on distributed message passing
How many GitHub stars does machinery have?
machinery has 7,971 GitHub stars in the directory's latest synchronization.
How do I install machinery?
Install machinery with the Go module system using `go get RichardKnop/machinery`. Check the repository for the current installation instructions.
What category does machinery belong to?
machinery is listed under Messaging, specifically Messaging.