# Work Queue (/sdks/elixir/how-to/work-queue)



## Overview [#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 [#prerequisites]

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

## Code [#code]

```elixir title="main.exs"
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 [#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 [#related]

* [Send & Receive](/sdks/elixir/tutorials/send-receive)
* [Poll Mode](/sdks/elixir/how-to/queues/poll-mode)