Command Group

Overview

A command consumer group turns a single command handler into a scalable worker pool: run multiple identical instances subscribed with the same group name, and the broker load-balances each incoming command to exactly one member instead of broadcasting it to all of them. This is how you add capacity to handle a growing command volume — start more workers in the same group — without changing anything on the caller's side.

Every subscriber passes the same group: alongside channel: on its CommandsSubscription to subscribe_to_commands; the broker tracks membership and picks one live member per command. send_command on the caller side is unaware groups exist — it just blocks for a result, which comes back from whichever worker happened to handle it via send_response.

Gotchas: group membership is scoped per channel — subscribers on the same channel with different group names each get their own full copy of every command (fan-out), which looks like a bug when you expected load-balancing. A slow handler still holds up the caller's timeout, since only one worker is ever picked. And if every member of the group is offline when a command arrives, the send simply fails or times out — commands aren't queued or replayed for a group that has no active listener.

Prerequisites

• KubeMQ server running on localhost:50000
• Ruby SDK installed (gem install kubemq)

Code

require 'kubemq'

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

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

  cancel = KubeMQ::CancellationToken.new

  sub1 = KubeMQ::CQ::CommandsSubscription.new(channel: channel, group: group)
  client.subscribe_to_commands(sub1, cancellation_token: cancel, on_error: lambda { |e|
    puts "Error: #{e.message}"
  }) do |cmd|
    puts "Worker-1 handling: #{cmd.metadata}"
    client.send_response(KubeMQ::CQ::CommandResponseMessage.new(
                           request_id: cmd.id, reply_channel: cmd.reply_channel, executed: true
                         ))
  end

  sub2 = KubeMQ::CQ::CommandsSubscription.new(channel: channel, group: group)
  client.subscribe_to_commands(sub2, cancellation_token: cancel, on_error: lambda { |e|
    puts "Error: #{e.message}"
  }) do |cmd|
    puts "Worker-2 handling: #{cmd.metadata}"
    client.send_response(KubeMQ::CQ::CommandResponseMessage.new(
                           request_id: cmd.id, reply_channel: cmd.reply_channel, executed: true
                         ))
  end
  sleep 1

  3.times do |i|
    msg = KubeMQ::CQ::CommandMessage.new(channel: channel, timeout: 10_000, metadata: "cmd-#{i}", body: "data-#{i}")
    result = client.send_command(msg)
    puts "Command #{i}: executed=#{result.executed}"
  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 CommandsSubscription enables load-balanced command routing.
• Each command is delivered to exactly one member of the group.
• Review timeouts, channel names, and client IDs before running against shared environments.

Related

• Pattern overview
• Ruby SDK Reference
• Send Command
• Query Group