KubeMQ
LearnQueuesScenarios

Background Job Worker

Distribute background jobs across a pool of workers using KubeMQ queues.

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.

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

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

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
}
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}")
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}`);
}
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());
}
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");
}
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")
}
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;
}
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(())
}
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
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

Worker Pool

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

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()
}
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()
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}`);
}
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);
    }
}
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);
    }
}
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)
    }
}
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));
    }
}
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?;
                }
            }
        }
    }
}
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
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

Advanced Configuration

Next Steps

Order Processing

Full pipeline with DLQ and retry.

Scheduled Tasks

Cron-like scheduling with delayed messages.

Was this page helpful?

Order Processing Pipeline

Build a reliable order processing pipeline with queues, DLQ, and retry logic.

Scheduled Task System

Implement cron-like scheduled tasks using KubeMQ delayed messages.

On this page

ArchitectureJob PublisherWorker PoolAdvanced ConfigurationNext Steps