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 that passes the same group: option to KubeMQ.Client.subscribe_to_events/3 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 call can flip between the two delivery models with one option.

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
• Elixir SDK installed ({:kubemq, "~> 1.0"} in mix.exs)

Code

channel = "elixir-events.consumer-group"
group = "my-consumer-group"

{:ok, client} = KubeMQ.Client.start_link(address: "localhost:50000", client_id: "elixir-events-group")
parent = self()

{:ok, sub1} =
  KubeMQ.Client.subscribe_to_events(client, channel,
    group: group,
    on_event: fn event ->
      IO.puts("[Sub-1] Got: #{event.body}")
      send(parent, {:sub1, event.body})
    end
  )

{:ok, sub2} =
  KubeMQ.Client.subscribe_to_events(client, channel,
    group: group,
    on_event: fn event ->
      IO.puts("[Sub-2] Got: #{event.body}")
      send(parent, {:sub2, event.body})
    end
  )

IO.puts("Two subscribers in group '#{group}' ready")
Process.sleep(500)

for i <- 1..6 do
  event = KubeMQ.Event.new(channel: channel, body: "Event #{i}")
  :ok = KubeMQ.Client.send_event(client, event)
end

IO.puts("Sent 6 events. Waiting for distribution...")
Process.sleep(2_000)

KubeMQ.Subscription.cancel(sub1)
KubeMQ.Subscription.cancel(sub2)
KubeMQ.Client.close(client)
IO.puts("Done. Each event was delivered to only one subscriber in the group.")

How It Works

• The group option assigns subscribers to a consumer group
• Each event is delivered to exactly one subscriber within the group
• This enables horizontal scaling of event processing workloads
• Without a group, each subscriber receives all events (fan-out)

Related

• Multiple Subscribers
• Events Pattern Overview