Architecture

An e-commerce platform submits orders to a queue. Worker processes pull orders, validate payment, update inventory, and send confirmations. Failed orders are retried with backoff, and permanently failed orders land in a dead letter queue for manual review.

Orders flow through a work queue to competing workers; failures retry, then dead-letter for review.

Order Submission

The API service creates an order and places it on the queue with DLQ policy and a 5-minute expiration.

order_api.go

type Order struct {
    OrderID   string  `json:"orderId"`
    Customer  string  `json:"customer"`
    Items     int     `json:"items"`
    Total     float64 `json:"total"`
    CreatedAt int64   `json:"createdAt"`
}

func submitOrder(ctx context.Context, client *kubemq.Client, order Order) error {
    body, _ := json.Marshal(order)
    msg := kubemq.NewQueueMessage().
        SetChannel("orders").
        SetBody(body).
        SetMetadata("order.created").
        SetTags(map[string]string{"customer": order.Customer}).
        SetMaxReceiveCount(5).
        SetMaxReceiveQueue("orders.dlq").
        SetExpirationSeconds(300)

    result, err := client.SendQueueMessage(ctx, msg)
    if err != nil {
        return err
    }
    log.Printf("Order %s submitted: id=%s", order.OrderID, result.MessageID)
    return nil
}

order_api.py

import json
import time

def submit_order(client, order):
    result = client.send_queue_message(
        QueueMessage(
            channel="orders",
            body=json.dumps(order).encode(),
            metadata="order.created",
            tags={"customer": order["customer"]},
            max_receive_count=5,
            max_receive_queue="orders.dlq",
            expiration_in_seconds=300,
        )
    )
    print(f"Order {order['orderId']} submitted: id={result.id}")
    return result

submit_order(client, {
    "orderId": "ORD-2001",
    "customer": "alice@example.com",
    "items": 3,
    "total": 149.97,
    "createdAt": int(time.time() * 1000),
})

order_api.ts

interface Order {
  orderId: string;
  customer: string;
  items: number;
  total: number;
  createdAt: number;
}

async function submitOrder(client: KubeMQClient, order: Order) {
  const result = await client.sendQueueMessage(
    createQueueMessage({
      channel: 'orders',
      body: JSON.stringify(order),
      metadata: 'order.created',
      tags: { customer: order.customer },
      policy: {
        maxReceiveCount: 5,
        maxReceiveQueue: 'orders.dlq',
        expirationSeconds: 300,
      },
    }),
  );
  console.log(`Order ${order.orderId} submitted: id=${result.messageId}`);
}

OrderApi.java

public void submitOrder(QueuesClient client, Order order) throws Exception {
    byte[] body = objectMapper.writeValueAsBytes(order);

    QueueMessage msg = QueueMessage.builder()
        .channel("orders")
        .body(body)
        .metadata("order.created")
        .tags(Map.of("customer", order.getCustomer()))
        .maxReceiveCount(5)
        .maxReceiveQueue("orders.dlq")
        .expirationSeconds(300)
        .build();

    SendQueueMessageResult result = client.sendQueueMessage(msg);
    System.out.printf("Order %s submitted: id=%s%n", order.getOrderId(), result.getMessageId());
}

OrderApi.cs

async Task SubmitOrder(QueuesClient client, Order order)
{
    var body = JsonSerializer.SerializeToUtf8Bytes(order);

    var result = await client.SendQueueMessageAsync(new QueueMessage
    {
        Channel = "orders",
        Body = body,
        Metadata = "order.created",
        Tags = new Dictionary<string, string> { ["customer"] = order.Customer },
        MaxReceiveCount = 5,
        MaxReceiveQueue = "orders.dlq",
        ExpirationSeconds = 300
    });

    Console.WriteLine($"Order {order.OrderId} submitted: id={result.MessageId}");
}

OrderApi.kt

suspend fun submitOrder(client: QueuesClient, order: Order) {
    val body = Json.encodeToString(order).toByteArray()

    val result = client.sendQueueMessage(QueueMessage(
        channel = "orders",
        body = body,
        metadata = "order.created",
        tags = mapOf("customer" to order.customer),
        maxReceiveCount = 5,
        maxReceiveQueue = "orders.dlq",
        expirationSeconds = 300
    ))

    println("Order ${order.orderId} submitted: id=${result.messageId}")
}

order_api.cpp

void submitOrder(kubemq::QueuesClient& client, const Order& order) {
    kubemq::QueueMessage msg;
    msg.channel = "orders";
    msg.body = order.toJson();
    msg.metadata = "order.created";
    msg.tags = {{"customer", order.customer}};
    msg.maxReceiveCount = 5;
    msg.maxReceiveQueue = "orders.dlq";
    msg.expirationSeconds = 300;

    auto result = client.sendQueueMessage(msg);
    std::cout << "Order " << order.orderId
              << " submitted: id=" << result.messageId << std::endl;
}

order_api.rs

use kubemq::prelude::*;
use kubemq::QueueMessageBuilder;
use std::collections::HashMap;

async fn submit_order(client: &KubemqClient, order: &Order) -> kubemq::Result<()> {
    let body = serde_json::to_vec(order).unwrap();

    let msg = QueueMessageBuilder::new()
        .channel("orders")
        .body(body)
        .metadata("order.created")
        .tags(HashMap::from([("customer".to_string(), order.customer.clone())]))
        .max_receive_count(5)
        .max_receive_queue("orders.dlq")
        .expiration_seconds(300)
        .build();

    let result = client.send_queue_message(msg).await?;
    println!("Order {} submitted: id={}", order.order_id, result.message_id);
    Ok(())
}

order_api.rb

require 'kubemq'
require 'json'

def submit_order(client, order)
  policy = KubeMQ::Queues::QueueMessagePolicy.new(
    max_receive_count: 5,
    max_receive_queue: 'orders.dlq',
    expiration_seconds: 300
  )
  msg = KubeMQ::Queues::QueueMessage.new(
    channel: 'orders',
    metadata: 'order.created',
    body: order.to_json,
    tags: { 'customer' => order['customer'] },
    policy: policy
  )
  result = client.send_queue_message(msg)
  puts "Order #{order['orderId']} submitted: id=#{result.id}"
end

order_api.exs

defp submit_order(client, order) do
  msg =
    KubeMQ.QueueMessage.new(
      channel: "orders",
      metadata: "order.created",
      body: Jason.encode!(order),
      tags: %{"customer" => order.customer},
      policy:
        KubeMQ.QueuePolicy.new(
          max_receive_count: 5,
          max_receive_queue: "orders.dlq",
          expiration_seconds: 300
        )
    )

  {:ok, result} = KubeMQ.Client.send_queue_message(client, msg)
  IO.puts("Order #{order.order_id} submitted: id=#{result.message_id}")
end

Order Worker

Workers poll for orders, process them, and acknowledge on success. Failed orders are nacked for retry.

order_worker.go

func runWorker(ctx context.Context, client *kubemq.Client, workerID string) {
    for {
        resp, err := client.PollQueue(ctx, &kubemq.PollRequest{
            Channel:            "orders",
            MaxItems:           5,
            WaitTimeoutSeconds: 10,
            VisibilitySeconds:  120,
        })
        if err != nil {
            log.Printf("[%s] Poll error: %v", workerID, err)
            time.Sleep(5 * time.Second)
            continue
        }

        for _, m := range resp.Messages {
            var order Order
            json.Unmarshal(m.Message.Body, &order)

            log.Printf("[%s] Processing order %s (attempt %d)",
                workerID, order.OrderID, m.Message.Attributes.ReceiveCount)

            if err := processOrder(order); err != nil {
                log.Printf("[%s] Failed: %v", workerID, err)
                continue
            }
            log.Printf("[%s] Order %s completed", workerID, order.OrderID)
        }
        resp.AckAll()
    }
}

order_worker.py

import json
import time

def run_worker(client, worker_id):
    while True:
        response = client.receive_queue_messages(
            channel="orders",
            max_messages=5,
            wait_timeout_in_seconds=10,
            visibility_seconds=120,
        )

        for msg in response.messages:
            order = json.loads(msg.body.decode("utf-8"))
            print(f"[{worker_id}] Processing order {order['orderId']} "
                  f"(attempt {msg.receive_count})")

            try:
                process_order(order)
                msg.ack()
                print(f"[{worker_id}] Order {order['orderId']} completed")
            except Exception as e:
                print(f"[{worker_id}] Failed: {e}")
                msg.nack()

        time.sleep(1)

order_worker.ts

async function runWorker(client: KubeMQClient, workerId: string) {
  while (true) {
    const messages = await client.receiveQueueMessages({
      channel: 'orders',
      maxMessages: 5,
      waitTimeoutSeconds: 10,
      visibilitySeconds: 120,
    });

    for (const msg of messages) {
      const order = JSON.parse(new TextDecoder().decode(msg.body));
      console.log(
        `[${workerId}] Processing order ${order.orderId} (attempt ${msg.receiveCount})`,
      );

      try {
        await processOrder(order);
        await msg.ack();
        console.log(`[${workerId}] Order ${order.orderId} completed`);
      } catch (err) {
        console.log(`[${workerId}] Failed:`, err);
        await msg.nack();
      }
    }

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

OrderWorker.java

public void runWorker(QueuesClient client, String workerId) {
    while (true) {
        ReceiveQueueMessagesResponse response = client.receiveQueueMessages(
            ReceiveQueueMessagesRequest.builder()
                .channel("orders").maxMessages(5)
                .waitTimeoutSeconds(10).visibilitySeconds(120).build());

        for (QueueMessageReceived msg : response.getMessages()) {
            Order order = objectMapper.readValue(msg.getBody(), Order.class);
            System.out.printf("[%s] Processing order %s (attempt %d)%n",
                workerId, order.getOrderId(), msg.getReceiveCount());

            try {
                processOrder(order);
                msg.ack();
                System.out.printf("[%s] Order %s completed%n", workerId, order.getOrderId());
            } catch (Exception e) {
                System.out.printf("[%s] Failed: %s%n", workerId, e.getMessage());
                msg.nack();
            }
        }
        Thread.sleep(1000);
    }
}

OrderWorker.cs

async Task RunWorker(QueuesClient client, string workerId)
{
    while (true)
    {
        var response = await client.ReceiveQueueMessagesAsync(new ReceiveQueueMessagesRequest
        {
            Channel = "orders", MaxMessages = 5,
            WaitTimeoutSeconds = 10, VisibilitySeconds = 120,
        });

        foreach (var msg in response.Messages)
        {
            var order = JsonSerializer.Deserialize<Order>(msg.Body.Span);
            Console.WriteLine($"[{workerId}] Processing {order.OrderId} (attempt {msg.ReceiveCount})");

            try
            {
                ProcessOrder(order);
                await msg.AckAsync();
                Console.WriteLine($"[{workerId}] Order {order.OrderId} completed");
            }
            catch (Exception ex)
            {
                Console.WriteLine($"[{workerId}] Failed: {ex.Message}");
                await msg.NAckAsync();
            }
        }
        await Task.Delay(1000);
    }
}

OrderWorker.kt

suspend fun runWorker(client: QueuesClient, workerId: String) {
    while (true) {
        val response = client.receiveQueueMessages(
            channel = "orders", maxMessages = 5,
            waitTimeoutSeconds = 10, visibilitySeconds = 120)

        for (msg in response.messages) {
            val order = Json.decodeFromString<Order>(String(msg.body))
            println("[$workerId] Processing ${order.orderId} (attempt ${msg.receiveCount})")

            try {
                processOrder(order)
                msg.ack()
                println("[$workerId] Order ${order.orderId} completed")
            } catch (e: Exception) {
                println("[$workerId] Failed: ${e.message}")
                msg.nack()
            }
        }
        delay(1000)
    }
}

order_worker.cpp

void runWorker(kubemq::QueuesClient& client, const std::string& workerId) {
    while (true) {
        auto response = client.receiveQueueMessages("orders", 5, 10, false, 120);

        for (const auto& msg : response.messages) {
            auto order = Order::fromJson(msg.body);
            std::cout << "[" << workerId << "] Processing " << order.orderId
                      << " (attempt " << msg.receiveCount << ")" << std::endl;

            try {
                processOrder(order);
                msg.ack();
                std::cout << "[" << workerId << "] Order " << order.orderId
                          << " completed" << std::endl;
            } catch (const std::exception& e) {
                std::cerr << "[" << workerId << "] Failed: " << e.what() << std::endl;
                msg.nack();
            }
        }
        std::this_thread::sleep_for(std::chrono::seconds(1));
    }
}

order_worker.rs

use kubemq::prelude::*;
use kubemq::PollRequest;
use std::time::Duration;

async fn run_worker(client: &KubemqClient, worker_id: &str) -> kubemq::Result<()> {
    let mut receiver = client.new_queue_downstream_receiver().await?;
    loop {
        let poll = PollRequest {
            channel: "orders".to_string(),
            max_items: 5,
            wait_timeout_seconds: 10,
            auto_ack: false,
        };
        let response = receiver.poll(poll).await?;

        for msg in &response.messages {
            let order: Order = serde_json::from_slice(&msg.message.body).unwrap();
            let attempt = msg.message.attributes.as_ref().map(|a| a.receive_count).unwrap_or(0);
            println!("[{}] Processing order {} (attempt {})", worker_id, order.order_id, attempt);

            match process_order(&order) {
                Ok(_) => {
                    msg.ack().await?;
                    println!("[{}] Order {} completed", worker_id, order.order_id);
                }
                Err(e) => {
                    println!("[{}] Failed: {}", worker_id, e);
                    msg.nack().await?;
                }
            }
        }
        tokio::time::sleep(Duration::from_secs(1)).await;
    }
}

order_worker.rb

require 'kubemq'
require 'json'

def run_worker(client, worker_id)
  receiver = client.create_downstream_receiver
  loop do
    request = KubeMQ::Queues::QueuePollRequest.new(
      channel: 'orders', max_items: 5, wait_timeout: 10
    )
    response = receiver.poll(request)

    response.messages.each do |m|
      order = JSON.parse(m.body)
      puts "[#{worker_id}] Processing order #{order['orderId']} (attempt #{m.attributes.receive_count})"

      begin
        process_order(order)
        m.ack
        puts "[#{worker_id}] Order #{order['orderId']} completed"
      rescue StandardError => e
        puts "[#{worker_id}] Failed: #{e.message}"
        m.nack
      end
    end
    sleep 1
  end
ensure
  receiver&.close
end

order_worker.exs

defp run_worker(client, worker_id) do
  case KubeMQ.Client.poll_queue(client, channel: "orders", max_items: 5, wait_timeout: 10_000) do
    {:ok, poll} ->
      # The Elixir SDK settles a poll batch as a whole: ack_all on success,
      # nack_all to return every message for redelivery (no per-message ack).
      results =
        Enum.map(poll.messages, fn msg ->
          order = Jason.decode!(msg.body)
          IO.puts("[#{worker_id}] Processing order #{order["orderId"]} (attempt #{msg.attributes.receive_count})")
          process_order(order)
        end)

      if Enum.all?(results, &(&1 == :ok)) do
        KubeMQ.PollResponse.ack_all(poll)
        IO.puts("[#{worker_id}] Batch completed")
      else
        KubeMQ.PollResponse.nack_all(poll)
        IO.puts("[#{worker_id}] Batch returned for retry")
      end

    {:error, err} ->
      IO.puts("[#{worker_id}] Poll error: #{err.message}")
  end

  Process.sleep(1_000)
  run_worker(client, worker_id)
end

Order Processing Pipeline

Architecture

Order Submission

Order Worker

Advanced Configuration

Next Steps

Background Workers

Retry with Backoff

On this page

Order Processing Pipeline

Architecture

Order Submission

Order Worker

Advanced Configuration

Idempotency

Visibility Tuning

DLQ Monitoring

Next Steps

Background Workers

Retry with Backoff

On this page