Task orchestration in Go Machinery.

Machinery is an asynchronous task queue/job queue based on distributed message passing. tasks/jobs are executed concurrently either by many workers on many servers or multiple worker processes on a single server using Golang’s goroutines. This uses distributed message passing so you can distribute tasks across many servers which allows you to scale horizontally

It is comparable in nature to task orchestration in Celery which is an outstanding Python library, in spite of the fact that Machinery has been planned from scale up and considering Golang’s qualities.

In order to illustrate better let’s implement task queuing in machinery in a real-world example.

we will build a web server that will accept a email , subject and body messages from client and enqueue the task to worker process ( aka machinery). And worker will consume the queue and send email asynchronously.

init the project

In this project we’ll use go module to manage dependencies. so let’s init go mod

go mod init github.com/<you>/go-machinery-example

Assuming our final project structure

Let’s start with a basic CLI app.

template for cli app

Lets implement server.StartServer and worker.StartWorker but before that we need to implement utils.GetMachineryServer() function to get the taskserver instance

we used Redis as our Broker and ResultBackend it’s actually the intermediate state where server and worker will exchange message or the task. you might also notice tasks.SendMail which registered with the taskserver. this is the actual task we will be dealing with.

Let’s make the task

in tasks.task.go we defined some functions to work with the task. Here DecodeToTask function decodes the base64 string to a payload struct. finally SendMail function to send a email. previously we already registered SendMail with machinery.

Run the server

in server.StartServer() we added a Http handler to accept a task from client that is just a json object of email,subject and body. And pass to taskserver through a task signature. that’s all , our task has been enqueued to the worker. worker will pick the task ASAP (if no delay settings)

Run the worker

in the worker side we simply recall GetMachineryServer to get the sametaskserver instance. To init a worker we need an unique tag and concurrency limit. here it’s 10

Finally, the usages !

go run main.go server
go run main.go worker

…

curl --request POST 'localhost:5000/send_task' --header 'Content-Type: application/json' --data-raw '{"email": "anyone@gmail.com","subject": "Hi","body": 

worker logs

Codebase

to get the complete code https://github.com/shemul/go-machinery