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: KubeMQ::CQ::QueriesSubscription.new(channel:, group:) passed to client.subscribe_to_queries load-balances across every subscriber sharing that channel and group. The sender calls client.send_query 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
Ruby SDK installed (gem install kubemq)

Code

main.rb

require 'kubemq'

address = ENV.fetch('KUBEMQ_ADDRESS', 'localhost:50000')
channel = 'ruby-rpc.query-group'
group = 'query-workers'

begin
  client = KubeMQ::CQClient.new(address: address, client_id: 'query-group-example')
  puts "Connected to #{address}"

  cancel = KubeMQ::CancellationToken.new

  sub1 = KubeMQ::CQ::QueriesSubscription.new(channel: channel, group: group)
  client.subscribe_to_queries(sub1, cancellation_token: cancel, on_error: lambda { |e|
    puts "Error: #{e.message}"
  }) do |query|
    puts "Worker-1 handling: #{query.metadata}"
    client.send_response(KubeMQ::CQ::QueryResponseMessage.new(
                           request_id: query.id, reply_channel: query.reply_channel,
                           executed: true, body: "worker-1-result"
                         ))
  end

  sub2 = KubeMQ::CQ::QueriesSubscription.new(channel: channel, group: group)
  client.subscribe_to_queries(sub2, cancellation_token: cancel, on_error: lambda { |e|
    puts "Error: #{e.message}"
  }) do |query|
    puts "Worker-2 handling: #{query.metadata}"
    client.send_response(KubeMQ::CQ::QueryResponseMessage.new(
                           request_id: query.id, reply_channel: query.reply_channel,
                           executed: true, body: "worker-2-result"
                         ))
  end
  sleep 1

  3.times do |i|
    msg = KubeMQ::CQ::QueryMessage.new(channel: channel, timeout: 10_000, metadata: "query-#{i}", body: "data-#{i}")
    result = client.send_query(msg)
    puts "Query #{i}: body=#{result.body}"
  end
rescue KubeMQ::Error => e
  puts "KubeMQ error: #{e.message}"
ensure
  cancel&.cancel
  client&.close
  puts 'Done'
end

How It Works

The group parameter on QueriesSubscription enables load-balanced query routing.
Each query is delivered to exactly one member of the group.
Review timeouts, channel names, and client IDs before running against shared environments.

Query Group

Overview

Prerequisites

Code

How It Works

On this page

Query Group

Overview

Prerequisites

Code

How It Works

Related

On this page