Work Queue

Overview

A work queue distributes a stream of tasks across a pool of workers so each task is handled exactly once, instead of every worker doing every task — the pattern you reach for whenever you need to parallelize processing (image resizing, batch jobs, background work) without coordinating which worker owns which item. The queue itself does that coordination: workers just keep polling, and the broker load-balances whatever is next in line across whichever workers happen to be asking.

poll_queue pulls a batch bounded by max_items and blocks up to wait_timeout if the queue is empty, so a worker long-polls instead of busy-looping or hanging forever. Delivery is competing-consumer: once one worker's poll call returns a message, no other worker gets it. KubeMQ.PollResponse.ack_all/1 settles the whole batch at once — until it's called, messages stay invisible and come back for redelivery if the worker never confirms, which is what makes the pattern at-least-once rather than fire-and-forget.

Gotchas: a worker that pulls a full max_items batch and then crashes before calling ack_all/1 leaves the whole batch to be redelivered — possibly to a different worker — so size batches to what you can safely redo. A short wait_timeout turns polling into a busy-loop that hammers the broker for empty results; too long delays workers noticing new work. And acking the whole batch together means a single bad message in the batch doesn't get isolated — a failure anywhere in processing should hold off ack_all/1 for the entire batch.

Prerequisites

• KubeMQ server running on localhost:50000
• Elixir SDK installed ({:kubemq, "~> 1.0"} in mix.exs)

Code

channel = "elixir-patterns.work-queue"
{:ok, client} = KubeMQ.Client.start_link(address: "localhost:50000", client_id: "elixir-work-queue")

IO.puts("Enqueueing 10 tasks...")

for i <- 1..10 do
  {:ok, _} = KubeMQ.Client.send_queue_message(client,
    KubeMQ.QueueMessage.new(
      channel: channel,
      body: ~s({"task_id": #{i}, "type": "process_image", "file": "img_#{i}.jpg"}),
      tags: %{"priority" => if(rem(i, 3) == 0, do: "high", else: "normal")}
    ))
end

IO.puts("All tasks enqueued")

workers =
  for worker_id <- 1..3 do
    Task.async(fn ->
      case KubeMQ.Client.poll_queue(client,
             channel: channel,
             max_items: 4,
             wait_timeout: 3_000
           ) do
        {:ok, poll} when length(poll.messages) > 0 ->
          Enum.each(poll.messages, fn msg ->
            IO.puts("[Worker-#{worker_id}] Processing: #{msg.body}")
            Process.sleep(100)
          end)

          {:ok, _} = KubeMQ.PollResponse.ack_all(poll)
          length(poll.messages)

        {:ok, _} ->
          0

        {:error, _} ->
          0
      end
    end)
  end

results = Task.await_many(workers, 15_000)
total = Enum.sum(results)
IO.puts("\nTotal processed: #{total} tasks across #{length(workers)} workers")

KubeMQ.Client.close(client)

How It Works

• 10 tasks are enqueued with tags for priority classification
• 3 concurrent workers poll the queue using Task.async/1
• Each worker polls up to 4 messages and processes them in order
• Task.await_many/2 waits for all workers to complete

Related

• Send & Receive
• Poll Mode