Batch Send

Overview

Batch send groups several queue messages into one call instead of sending them one at a time. Reach for it when publishing many related items together — importing records, fanning out a set of jobs, replaying a backlog — since sending each message individually pays a full round trip per message, while batching amortizes that cost across the whole set.

It works by building a slice of *kubemq.QueueMessage values with kubemq.NewQueueMessage().SetChannel(...).SetBody(...), then passing the slice to client.SendQueueMessages(ctx, batch) in a single RPC. The broker enqueues each message independently and returns a []QueueMessageSendResult in the same order as the input, one result per message.

Gotchas: batching isn't atomic — the broker can accept some messages and reject others in the same call, so always check IsError on every result instead of trusting an overall success; a batch is still one bounded request, so it doesn't help continuous, open-ended publishing (use a persistent upstream stream for that); and very large batches raise the size and latency of that single call, so there's a practical ceiling before splitting into multiple batches pays off.

Prerequisites

• KubeMQ server running on localhost:50000
• Go SDK installed (go get github.com/kubemq-io/kubemq-go/v2)

Code

// Example: queues/batch-send
//
// Demonstrates sending multiple queue messages in a single batch operation.
// Batch send reduces round trips for high-throughput scenarios.
//
// Channel: go-queues.batch-send
// Client ID: go-queues-batch-send-client
//
// Run with a KubeMQ server on localhost:50000
// (see https://docs.kubemq.io/deploy).
package main

import (
	"context"
	"fmt"
	"log"
	"time"

	"github.com/kubemq-io/kubemq-go/v2"
)

func main() {
	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
	defer cancel()

	client, err := kubemq.NewClient(ctx,
		kubemq.WithAddress("localhost", 50000),
		kubemq.WithClientId("go-queues-batch-send-client"),
	)
	if err != nil {
		log.Fatal(err)
	}
	defer client.Close()

	channel := "go-queues.batch-send"

	// Create a batch of queue messages.
	batch := []*kubemq.QueueMessage{
		kubemq.NewQueueMessage().SetChannel(channel).SetBody([]byte("batch-msg-1")),
		kubemq.NewQueueMessage().SetChannel(channel).SetBody([]byte("batch-msg-2")),
		kubemq.NewQueueMessage().SetChannel(channel).SetBody([]byte("batch-msg-3")),
	}

	// Send all messages in a single batch operation.
	results, err := client.SendQueueMessages(ctx, batch)
	if err != nil {
		log.Fatal(err)
	}
	for i, r := range results {
		fmt.Printf("Batch[%d]: id=%s error=%v\n", i, r.MessageID, r.IsError)
	}
	fmt.Println("Batch send complete")
}

How It Works

1. A slice of *kubemq.QueueMessage values is built using the fluent kubemq.NewQueueMessage().SetChannel(...).SetBody(...) builder; all messages target the same channel.
2. client.SendQueueMessages(ctx, batch) sends the entire slice in a single RPC call — one round trip to the broker regardless of batch size, which lowers per-message overhead at high volumes.
3. The returned []QueueMessageSendResult has one entry per input message; check r.IsError on each to detect partial failures where some messages are enqueued but others are not.
4. For higher-throughput scenarios where multiple batches need to be sent without waiting for each round trip, see the Stream Send example which uses a persistent upstream stream.

Related

• Pattern overview
• Go SDK Reference
• Send & Receive
• Ack All