# Background Job Worker (/learn/queues/scenarios/background-workers)



## Architecture [#architecture]

A job publisher enqueues tasks (image processing, report generation, email sending). A pool of workers competes for jobs — each job is processed by exactly one worker. Workers acknowledge jobs on success and nack on failure for automatic retry.

<Mermaid
  chart="graph LR
  APP[&#x22;Job Publisher&#x22;]
  JQ[[&#x22;background-jobs<br/>queue&#x22;]]
  W1[&#x22;Worker 1&#x22;]
  W2[&#x22;Worker 2&#x22;]
  W3[&#x22;Worker 3&#x22;]
  DONE[&#x22;Job complete&#x22;]

  APP -- enqueue --> JQ
  JQ -- compete --> W1
  JQ -- compete --> W2
  JQ -- compete --> W3
  W1 -- ack --> DONE
  W2 -. nack / retry .-> JQ

  class APP,W1,W2,W3 client
  class JQ queue
  class DONE data"
/>

*A pool of workers competes for jobs on a single queue — each job is delivered to exactly one worker, which acks on success or nacks to requeue for retry.*

## Job Publisher [#job-publisher]

Enqueue background jobs with metadata describing the job type and priority via tags.

<Tabs groupId="language" items="['Go', 'Python', 'Node.js', 'Java', 'C#', 'Kotlin', 'C++', 'Rust', 'Ruby', 'Elixir']">
  <Tab value="Go">
    ```go title="job_publisher.go"
    type Job struct {
        JobID   string `json:"jobId"`
        Type    string `json:"type"`
        Payload string `json:"payload"`
    }

    func enqueueJob(ctx context.Context, client *kubemq.Client, job Job) error {
        body, _ := json.Marshal(job)
        msg := kubemq.NewQueueMessage().
            SetChannel("background-jobs").
            SetBody(body).
            SetMetadata(job.Type).
            SetTags(map[string]string{
                "type":     job.Type,
                "priority": "normal",
            }).
            SetMaxReceiveCount(3).
            SetMaxReceiveQueue("background-jobs.dlq")

        _, err := client.SendQueueMessage(ctx, msg)
        return err
    }
    ```
  </Tab>

  <Tab value="Python">
    ```python title="job_publisher.py"
    import json

    def enqueue_job(client, job):
        result = client.send_queue_message(
            QueueMessage(
                channel="background-jobs",
                body=json.dumps(job).encode(),
                metadata=job["type"],
                tags={"type": job["type"], "priority": "normal"},
                max_receive_count=3,
                max_receive_queue="background-jobs.dlq",
            )
        )
        print(f"Job {job['jobId']} enqueued: id={result.id}")
    ```
  </Tab>

  <Tab value="Node.js">
    ```typescript title="job_publisher.ts"
    async function enqueueJob(client: KubeMQClient, job: Job) {
      const result = await client.sendQueueMessage(
        createQueueMessage({
          channel: 'background-jobs',
          body: JSON.stringify(job),
          metadata: job.type,
          tags: { type: job.type, priority: 'normal' },
          policy: { maxReceiveCount: 3, maxReceiveQueue: 'background-jobs.dlq' },
        }),
      );
      console.log(`Job ${job.jobId} enqueued: id=${result.messageId}`);
    }
    ```
  </Tab>

  <Tab value="Java">
    ```java title="JobPublisher.java"
    public void enqueueJob(QueuesClient client, Job job) throws Exception {
        QueueMessage msg = QueueMessage.builder()
            .channel("background-jobs")
            .body(objectMapper.writeValueAsBytes(job))
            .metadata(job.getType())
            .tags(Map.of("type", job.getType(), "priority", "normal"))
            .maxReceiveCount(3)
            .maxReceiveQueue("background-jobs.dlq")
            .build();

        client.sendQueueMessage(msg);
        System.out.printf("Job %s enqueued%n", job.getJobId());
    }
    ```
  </Tab>

  <Tab value="C#">
    ```csharp title="JobPublisher.cs"
    async Task EnqueueJob(QueuesClient client, Job job)
    {
        await client.SendQueueMessageAsync(new QueueMessage
        {
            Channel = "background-jobs",
            Body = JsonSerializer.SerializeToUtf8Bytes(job),
            Metadata = job.Type,
            Tags = new Dictionary<string, string> { ["type"] = job.Type, ["priority"] = "normal" },
            MaxReceiveCount = 3,
            MaxReceiveQueue = "background-jobs.dlq"
        });
        Console.WriteLine($"Job {job.JobId} enqueued");
    }
    ```
  </Tab>

  <Tab value="Kotlin">
    ```kotlin title="JobPublisher.kt"
    suspend fun enqueueJob(client: QueuesClient, job: Job) {
        client.sendQueueMessage(QueueMessage(
            channel = "background-jobs",
            body = Json.encodeToString(job).toByteArray(),
            metadata = job.type,
            tags = mapOf("type" to job.type, "priority" to "normal"),
            maxReceiveCount = 3,
            maxReceiveQueue = "background-jobs.dlq"
        ))
        println("Job ${job.jobId} enqueued")
    }
    ```
  </Tab>

  <Tab value="C++">
    ```cpp title="job_publisher.cpp"
    void enqueueJob(kubemq::QueuesClient& client, const Job& job) {
        kubemq::QueueMessage msg;
        msg.channel = "background-jobs";
        msg.body = job.toJson();
        msg.metadata = job.type;
        msg.tags = {{"type", job.type}, {"priority", "normal"}};
        msg.maxReceiveCount = 3;
        msg.maxReceiveQueue = "background-jobs.dlq";

        client.sendQueueMessage(msg);
        std::cout << "Job " << job.jobId << " enqueued" << std::endl;
    }
    ```
  </Tab>

  <Tab value="Rust">
    ```rust title="job_publisher.rs"
    use kubemq::prelude::*;
    use kubemq::QueueMessageBuilder;
    use std::collections::HashMap;

    async fn enqueue_job(client: &KubemqClient, job: &Job) -> kubemq::Result<()> {
        let mut tags = HashMap::new();
        tags.insert("type".to_string(), job.job_type.clone());
        tags.insert("priority".to_string(), "normal".to_string());

        let msg = QueueMessageBuilder::new()
            .channel("background-jobs")
            .body(serde_json::to_vec(job).unwrap())
            .metadata(job.job_type.clone())
            .tags(tags)
            .max_receive_count(3)
            .max_receive_queue("background-jobs.dlq")
            .build();

        let result = client.send_queue_message(msg).await?;
        println!("Job {} enqueued: id={}", job.job_id, result.message_id);
        Ok(())
    }
    ```
  </Tab>

  <Tab value="Ruby">
    ```ruby title="job_publisher.rb"
    require 'kubemq'
    require 'json'

    def enqueue_job(client, job)
      policy = KubeMQ::Queues::QueueMessagePolicy.new(
        max_receive_count: 3,
        max_receive_queue: 'background-jobs.dlq'
      )
      msg = KubeMQ::Queues::QueueMessage.new(
        channel: 'background-jobs',
        body: job.to_json,
        metadata: job['type'],
        tags: { 'type' => job['type'], 'priority' => 'normal' },
        policy: policy
      )
      result = client.send_queue_message(msg)
      puts "Job #{job['jobId']} enqueued: id=#{result.id}"
    end
    ```
  </Tab>

  <Tab value="Elixir">
    ```elixir title="job_publisher.exs"
    defmodule JobPublisher do
      def enqueue_job(client, job) do
        msg = KubeMQ.QueueMessage.new(
          channel: "background-jobs",
          body: Jason.encode!(job),
          metadata: job["type"],
          tags: %{"type" => job["type"], "priority" => "normal"},
          policy: KubeMQ.QueuePolicy.new(
            max_receive_count: 3,
            max_receive_queue: "background-jobs.dlq"
          )
        )

        {:ok, result} = KubeMQ.Client.send_queue_message(client, msg)
        IO.puts("Job #{job["jobId"]} enqueued: id=#{result.message_id}")
      end
    end
    ```
  </Tab>
</Tabs>

## Worker Pool [#worker-pool]

Each worker runs in a loop, polling for jobs in batches. KubeMQ ensures each job goes to exactly one worker.

<Tabs groupId="language" items="['Go', 'Python', 'Node.js', 'Java', 'C#', 'Kotlin', 'C++', 'Rust', 'Ruby', 'Elixir']">
  <Tab value="Go">
    ```go title="worker_pool.go"
    func startWorker(ctx context.Context, client *kubemq.Client, id int) {
        workerID := fmt.Sprintf("worker-%d", id)
        log.Printf("[%s] Started", workerID)

        for {
            select {
            case <-ctx.Done():
                return
            default:
            }

            resp, err := client.PollQueue(ctx, &kubemq.PollRequest{
                Channel:            "background-jobs",
                MaxItems:           5,
                WaitTimeoutSeconds: 10,
                VisibilitySeconds:  300,
            })
            if err != nil {
                log.Printf("[%s] Poll error: %v", workerID, err)
                time.Sleep(5 * time.Second)
                continue
            }

            for _, m := range resp.Messages {
                var job Job
                json.Unmarshal(m.Message.Body, &job)
                log.Printf("[%s] Processing job %s (%s)", workerID, job.JobID, job.Type)

                if err := executeJob(job); err != nil {
                    log.Printf("[%s] Job %s failed: %v", workerID, job.JobID, err)
                } else {
                    log.Printf("[%s] Job %s completed", workerID, job.JobID)
                }
            }
            resp.AckAll()
        }
    }

    func main() {
        ctx := context.Background()
        client, _ := kubemq.NewClient(ctx,
            kubemq.WithAddress("localhost", 50000))
        defer client.Close()

        for i := 1; i <= 3; i++ {
            go startWorker(ctx, client, i)
        }
        <-ctx.Done()
    }
    ```
  </Tab>

  <Tab value="Python">
    ```python title="worker_pool.py"
    import json
    import threading
    import time

    def start_worker(client, worker_id):
        print(f"[{worker_id}] Started")
        while True:
            response = client.receive_queue_messages(
                channel="background-jobs",
                max_messages=5,
                wait_timeout_in_seconds=10,
                visibility_seconds=300,
            )

            for msg in response.messages:
                job = json.loads(msg.body.decode("utf-8"))
                print(f"[{worker_id}] Processing job {job['jobId']} ({job['type']})")

                try:
                    execute_job(job)
                    msg.ack()
                    print(f"[{worker_id}] Job {job['jobId']} completed")
                except Exception as e:
                    print(f"[{worker_id}] Job {job['jobId']} failed: {e}")
                    msg.nack()

            time.sleep(1)

    for i in range(1, 4):
        threading.Thread(target=start_worker, args=(client, f"worker-{i}"), daemon=True).start()
    ```
  </Tab>

  <Tab value="Node.js">
    ```typescript title="worker_pool.ts"
    async function startWorker(client: KubeMQClient, workerId: string) {
      console.log(`[${workerId}] Started`);

      while (true) {
        const messages = await client.receiveQueueMessages({
          channel: 'background-jobs',
          maxMessages: 5,
          waitTimeoutSeconds: 10,
          visibilitySeconds: 300,
        });

        for (const msg of messages) {
          const job = JSON.parse(new TextDecoder().decode(msg.body));
          console.log(`[${workerId}] Processing job ${job.jobId} (${job.type})`);

          try {
            await executeJob(job);
            await msg.ack();
            console.log(`[${workerId}] Job ${job.jobId} completed`);
          } catch (err) {
            console.log(`[${workerId}] Job ${job.jobId} failed:`, err);
            await msg.nack();
          }
        }

        await new Promise((r) => setTimeout(r, 1000));
      }
    }

    for (let i = 1; i <= 3; i++) {
      startWorker(client, `worker-${i}`);
    }
    ```
  </Tab>

  <Tab value="Java">
    ```java title="WorkerPool.java"
    public void startWorker(QueuesClient client, String workerId) {
        System.out.printf("[%s] Started%n", workerId);

        while (true) {
            ReceiveQueueMessagesResponse response = client.receiveQueueMessages(
                ReceiveQueueMessagesRequest.builder()
                    .channel("background-jobs").maxMessages(5)
                    .waitTimeoutSeconds(10).visibilitySeconds(300).build());

            for (QueueMessageReceived msg : response.getMessages()) {
                Job job = objectMapper.readValue(msg.getBody(), Job.class);
                System.out.printf("[%s] Processing job %s (%s)%n",
                    workerId, job.getJobId(), job.getType());

                try {
                    executeJob(job);
                    msg.ack();
                } catch (Exception e) {
                    System.out.printf("[%s] Job %s failed: %s%n", workerId, job.getJobId(), e);
                    msg.nack();
                }
            }
            Thread.sleep(1000);
        }
    }
    ```
  </Tab>

  <Tab value="C#">
    ```csharp title="WorkerPool.cs"
    async Task StartWorker(QueuesClient client, string workerId)
    {
        Console.WriteLine($"[{workerId}] Started");

        while (true)
        {
            var response = await client.ReceiveQueueMessagesAsync(new ReceiveQueueMessagesRequest
            {
                Channel = "background-jobs", MaxMessages = 5,
                WaitTimeoutSeconds = 10, VisibilitySeconds = 300,
            });

            foreach (var msg in response.Messages)
            {
                var job = JsonSerializer.Deserialize<Job>(msg.Body.Span);
                Console.WriteLine($"[{workerId}] Processing job {job.JobId} ({job.Type})");

                try
                {
                    ExecuteJob(job);
                    await msg.AckAsync();
                }
                catch (Exception ex)
                {
                    Console.WriteLine($"[{workerId}] Job {job.JobId} failed: {ex.Message}");
                    await msg.NAckAsync();
                }
            }
            await Task.Delay(1000);
        }
    }
    ```
  </Tab>

  <Tab value="Kotlin">
    ```kotlin title="WorkerPool.kt"
    suspend fun startWorker(client: QueuesClient, workerId: String) {
        println("[$workerId] Started")

        while (true) {
            val response = client.receiveQueueMessages(
                channel = "background-jobs", maxMessages = 5,
                waitTimeoutSeconds = 10, visibilitySeconds = 300)

            for (msg in response.messages) {
                val job = Json.decodeFromString<Job>(String(msg.body))
                println("[$workerId] Processing job ${job.jobId} (${job.type})")

                try {
                    executeJob(job)
                    msg.ack()
                } catch (e: Exception) {
                    println("[$workerId] Job ${job.jobId} failed: ${e.message}")
                    msg.nack()
                }
            }
            delay(1000)
        }
    }
    ```
  </Tab>

  <Tab value="C++">
    ```cpp title="worker_pool.cpp"
    void startWorker(kubemq::QueuesClient& client, const std::string& workerId) {
        std::cout << "[" << workerId << "] Started" << std::endl;

        while (true) {
            auto response = client.receiveQueueMessages(
                "background-jobs", 5, 10, false, 300);

            for (const auto& msg : response.messages) {
                auto job = Job::fromJson(msg.body);
                std::cout << "[" << workerId << "] Processing job "
                          << job.jobId << " (" << job.type << ")" << std::endl;

                try {
                    executeJob(job);
                    msg.ack();
                } catch (const std::exception& e) {
                    std::cerr << "[" << workerId << "] Failed: " << e.what() << std::endl;
                    msg.nack();
                }
            }
            std::this_thread::sleep_for(std::chrono::seconds(1));
        }
    }
    ```
  </Tab>

  <Tab value="Rust">
    ```rust title="worker_pool.rs"
    use kubemq::prelude::*;
    use kubemq::PollRequest;

    async fn start_worker(client: &KubemqClient, worker_id: &str) -> kubemq::Result<()> {
        println!("[{}] Started", worker_id);
        let mut receiver = client.new_queue_downstream_receiver().await?;

        loop {
            let poll = PollRequest {
                channel: "background-jobs".to_string(),
                max_items: 5,
                wait_timeout_seconds: 10,
                auto_ack: false,
            };
            let response = receiver.poll(poll).await?;

            for msg in &response.messages {
                let job: Job = serde_json::from_slice(&msg.message.body).unwrap();
                println!("[{}] Processing job {} ({})", worker_id, job.job_id, job.job_type);

                match execute_job(&job) {
                    Ok(_) => {
                        msg.ack().await?;
                        println!("[{}] Job {} completed", worker_id, job.job_id);
                    }
                    Err(e) => {
                        println!("[{}] Job {} failed: {}", worker_id, job.job_id, e);
                        msg.nack().await?;
                    }
                }
            }
        }
    }
    ```
  </Tab>

  <Tab value="Ruby">
    ```ruby title="worker_pool.rb"
    require 'kubemq'
    require 'json'

    def start_worker(client, worker_id)
      puts "[#{worker_id}] Started"
      receiver = client.create_downstream_receiver

      loop do
        request = KubeMQ::Queues::QueuePollRequest.new(
          channel: 'background-jobs',
          max_items: 5,
          wait_timeout: 10
        )
        response = receiver.poll(request)
        next if response.error?

        response.messages.each do |msg|
          job = JSON.parse(msg.body)
          puts "[#{worker_id}] Processing job #{job['jobId']} (#{job['type']})"

          begin
            execute_job(job)
            msg.ack
            puts "[#{worker_id}] Job #{job['jobId']} completed"
          rescue StandardError => e
            puts "[#{worker_id}] Job #{job['jobId']} failed: #{e.message}"
            msg.nack
          end
        end
      end
    ensure
      receiver&.close
    end
    ```
  </Tab>

  <Tab value="Elixir">
    ```elixir title="worker_pool.exs"
    defmodule WorkerPool do
      # Elixir acks/nacks by sequence range on the poll transaction
      # rather than per individual message.
      def start_worker(client, worker_id) do
        IO.puts("[#{worker_id}] Started")

        case KubeMQ.Client.poll_queue(client,
               channel: "background-jobs",
               max_items: 5,
               wait_timeout: 10_000
             ) do
          {:ok, poll} when length(poll.messages) > 0 ->
            {ack_seqs, nack_seqs} =
              Enum.reduce(poll.messages, {[], []}, fn msg, {ok, fail} ->
                job = Jason.decode!(msg.body)
                seq = msg.attributes.sequence
                IO.puts("[#{worker_id}] Processing job #{job["jobId"]} (#{job["type"]})")

                case execute_job(job) do
                  :ok -> {[seq | ok], fail}
                  {:error, _} -> {ok, [seq | fail]}
                end
              end)

            if ack_seqs != [], do: KubeMQ.PollResponse.ack_range(poll, ack_seqs)
            if nack_seqs != [], do: KubeMQ.PollResponse.nack_range(poll, nack_seqs)
            start_worker(client, worker_id)

          _ ->
            start_worker(client, worker_id)
        end
      end
    end
    ```
  </Tab>
</Tabs>

## Advanced Configuration [#advanced-configuration]

<Accordions>
  <Accordion title="Batch Receive for Throughput">
    Increase `maxMessages` to pull more jobs per poll cycle. Workers process jobs in batches, reducing network round trips.
  </Accordion>

  <Accordion title="Priority Queues">
    Use separate channels for different priorities (`jobs.high`, `jobs.normal`, `jobs.low`). Workers poll high-priority channels first.
  </Accordion>

  <Accordion title="Job TTL">
    Set `expirationSeconds` on time-sensitive jobs. Stale jobs are automatically discarded instead of consuming worker capacity.
  </Accordion>
</Accordions>

## Next Steps [#next-steps]

<Cards>
  <Card title="Order Processing" href="/learn/queues/scenarios/order-processing" description="Full pipeline with DLQ and retry." />

  <Card title="Scheduled Tasks" href="/learn/queues/scenarios/scheduled-tasks" description="Cron-like scheduling with delayed messages." />
</Cards>