Query Group

Overview

A consumer group scales query handling horizontally without touching the caller's side. Instead of one process answering every query on a channel, you run several identical handler instances under the same group name, and the broker routes each query to exactly one member — never to all of them. That turns a single responder into a pool you can grow or shrink to match load, which matters for anything RPC-shaped: a lookup service, a cache-fill handler, a synchronous read path behind an API.

It works by tying group membership to the subscription: client.SubscribeToQueries(ctx, channel, group, ...) with a non-empty group load-balances across every subscriber sharing that channel and group. The sender calls client.SendQuery exactly as it would against a single handler — it never knows how many members exist or which one answered.

Gotchas: channel and group name must match exactly, or a typo quietly creates a second, empty group instead of erroring. Omit group and every subscriber reverts to broadcast, each answering independently. A stuck group member isn't bypassed — the caller just sees a timeout.

Prerequisites

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

Code

// Example: queries/consumer-group
//
// Demonstrates load-balanced query handling with consumer groups.
// Multiple handlers in the same group share the query workload.
//
// Channel: go-queries.consumer-group
// Client ID: go-queries-consumer-group-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-queries-consumer-group-client"),
	)
	if err != nil {
		log.Fatal(err)
	}
	defer client.Close()

	channel := "go-queries.consumer-group"
	group := "go-queries-reader-group"
	done := make(chan struct{})

	// Subscribe with a consumer group.
	sub, err := client.SubscribeToQueries(ctx, channel, group,
		kubemq.WithOnQueryReceive(func(q *kubemq.QueryReceive) {
			fmt.Printf("Worker received: body=%s\n", q.Body)
			resp := kubemq.NewQueryReply().
				SetRequestId(q.Id).
				SetResponseTo(q.ResponseTo).
				SetBody([]byte("group-result")).
				SetExecutedAt(time.Now())
			_ = client.SendQueryResponse(ctx, resp)
			close(done)
		}),
		kubemq.WithOnError(func(err error) {
			log.Println("Group error:", err)
		}),
	)
	if err != nil {
		log.Fatal(err)
	}
	defer sub.Unsubscribe()
	time.Sleep(300 * time.Millisecond)

	// Send a query to the group.
	qResp, err := client.SendQuery(ctx, kubemq.NewQuery().
		SetChannel(channel).
		SetBody([]byte("group-fetch")).
		SetTimeout(10*time.Second))
	if err != nil {
		log.Fatal(err)
	}
	fmt.Printf("Group response: executed=%v body=%s\n", qResp.Executed, qResp.Body)

	<-done
}

How It Works

1. client.SubscribeToQueries(ctx, channel, group, opts...) passes group := "go-queries-reader-group", so the broker load-balances incoming queries across all subscribers in the group.
2. When multiple instances subscribe with the same channel and group, each query is handled by exactly one instance — suitable for scaling read-heavy services.
3. client.SendQuery blocks until one group member responds with a QueryReply; the response body is available in qResp.Body.
4. defer sub.Unsubscribe() removes this subscriber from the group at exit; remaining group members continue to handle new queries without interruption.

Related

• Pattern overview
• Go SDK Reference
• Send Query
• Handle Query