# Query Group (/sdks/go/how-to/rpc/query-group)



## Overview [#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 [#prerequisites]

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

## Code [#code]

```go title="main.go"
// 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 [#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 [#related]

* [Pattern overview](/learn/rpc/getting-started)
* [Go SDK Reference](/sdks/go/reference)
* [Send Query](/sdks/go/tutorials/query-send)
* [Handle Query](/sdks/go/how-to/rpc/query-handle)