Consumer Group

Overview

A consumer group turns Events pub/sub from a broadcast into a work queue. By default every subscriber on a channel gets every event — fine for notifications, but wasteful when you want a pool of workers to split a stream of tasks so each one is handled exactly once. Reach for a consumer group whenever you're scaling out event processing and duplicate work isn't just wasteful but actively wrong (double-charging a customer, double-sending an alert).

It works by naming a group when you subscribe: every subscriber whose EventsSubscription.new sets the same group: value (passed to subscribe_to_events) joins that group, and the broker round-robins each event to exactly one member instead of fanning it out to all of them. Omitting group: reverts to normal fan-out, so the same subscription shape can flip between the two delivery models with one keyword argument.

Gotchas: consumer groups are scoped per channel — subscribing to the same group on a different channel does not share load balancing across channels. A group with zero active subscribers behaves like no subscribers at all; events aren't queued for a group that's temporarily empty the way they are for durable queue messages. And because delivery is round-robin rather than content-aware, you can't route specific events to specific workers within a group — if you need that, partition by channel instead.

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-events.consumer-group'
group = 'worker-group'

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

  cancel = KubeMQ::CancellationToken.new

  sub1 = KubeMQ::PubSub::EventsSubscription.new(channel: channel, group: group)
  client.subscribe_to_events(sub1, cancellation_token: cancel, on_error: lambda { |e|
    puts "Error: #{e.message}"
  }) do |event|
    puts "Worker-1 received: #{event.metadata}"
  end

  sub2 = KubeMQ::PubSub::EventsSubscription.new(channel: channel, group: group)
  client.subscribe_to_events(sub2, cancellation_token: cancel, on_error: lambda { |e|
    puts "Error: #{e.message}"
  }) do |event|
    puts "Worker-2 received: #{event.metadata}"
  end
  sleep 1

  5.times do |i|
    msg = KubeMQ::PubSub::EventMessage.new(channel: channel, metadata: "event #{i}", body: "data-#{i}")
    client.send_event(msg)
    puts "Sent event #{i}"
  end

  sleep 2
rescue KubeMQ::Error => e
  puts "KubeMQ error: #{e.message}"
ensure
  cancel&.cancel
  client&.close
  puts 'Done'
end

How It Works

• When multiple subscribers share the same group, the broker delivers each event to only one member of the group.
• This enables horizontal scaling of event processing — add more workers to increase throughput.
• Without a group, all subscribers receive every event (fan-out behavior).
• Review timeouts, channel names, and client IDs before running against shared environments.

Related

• Pattern overview
• Ruby SDK Reference
• Basic Pub/Sub
• Multiple Subscribers