Pooling and queue libraries in Elixir

I’m not aware of an embedded solution for a distributed queue. I’d likely try RabbitMQ in that scenario.

However, if you want a local persistent queue with absolutely no external dependencies, then the solution would not be overly complex. I don’t know if there are libraries which do this out of the box, but rolling your own simple solution would IMO not be too complicated, and is at least a nice exercise of a bit of OTP

The idea is to have an in-memory queue, and persist the queue state to disk every time an item is added, and every time the job succeeds.

Persisting without an external component can be done in a couple of ways:

• File.write + :erlang.term_to_binary
• dets
• mnesia

In a simplest solution, you’d have one process, a GenServer, which would act as a scheduler. The scheduler reads from the queue file, and starts the first job in a separate process. If the job succeeds, the scheduler removes it, persists the new queue state, and starts the next job, if there is one. The scheduler should also monitor the job process (or link to it), to detect if it crashed. Then you need to handle the crash by retrying, deferring the job, giving up.

Every time you want to add another item, you send a request to the scheduler, which adds it to the end of the in memory structure (for example :queue), stores to disk, and starts the job if nothing is running.

I’d suggest to try rolling a simple implementation, perhaps starting with a plain in-memory queue, and then adding naive persistence with File.write(:erlang.term_to_binary(queue)).

If you have more elaborate needs, such as pooled or rate-limited queues, you could take a look at jobs. I’m actually using it myself for these particular cases, and I’m quite happy with it.

The author of Sidekiq has a new project for async work queues called Faktory which is language agnostic and addresses some of the main design short comings of Sidekiq.

There is a fully featured Elixir client for it that is actively being developed:

I don’t know of any. But you don’t need a wrapper, you can use jobs directly. The examples in Erlang are here, and the comparison of syntax differences can be found here. So e.g. if we take this example:

jobs:add_queue(q, [{standard_rate,1}]).
...
jobs:run(q, fun() -> io:fwrite("job: ~p~n", [time()]) end).

It could be roughly translated as e.g.:

:jobs.add_queue(:q, [standard_rate: 1])
...
:jobs.run(:q, fn -> IO.inspect(DateTime.utc_now()) end)

More generally, to use jobs, you need to first create a named queue with :jobs.add_queue(queue_name, opts), where queue_name is an atom. Typically this can be done in the application start callback.

Then you can run a queued operation from any process as :jobs.run(queue_name, fn -> ... end).

Take a look ad add_queue docs, in especially the “regulators” section. To create a rate-limited queue, you can pass rate: [...]. To create a counter queue (limit max simultaneous jobs), you can pass counter: [...].