Consumer Group
Load-balanced event delivery using consumer groups
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 name as the second argument to SubscribeToEvents() joins that group, and the broker round-robins each event to exactly one member instead of fanning it out to all of them. Passing an empty string reverts to normal fan-out, so the same call can flip between the two delivery models with one 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 - C++ SDK installed (vcpkg or CMake FetchContent)
- C++17 compiler (GCC 9+, Clang 9+, MSVC 2019+)
Code
// Example: events/consumer_group
//
// Demonstrates load-balanced event consumption using consumer groups.
// When multiple subscribers share the same group on the same channel,
// each event is delivered to exactly one subscriber in the group.
//
// Channel: cpp-events.consumer-group
// Client ID: cpp-events-consumer-group-client
//
// Run with a KubeMQ server on localhost:50000
// (see https://docs.kubemq.io/deploy).
#include <kubemq/kubemq.h>
#include <chrono>
#include <iostream>
#include <thread>
int main() {
std::cout << "[1] Connecting to localhost:50000" << std::endl;
kubemq::ClientOptions options;
options.set_address("localhost", 50000);
options.set_client_id("cpp-events-consumer-group-client");
auto client_result = kubemq::Client::Create(options);
if (!client_result.ok()) {
std::cerr << "[ERROR] Failed to create client: " << client_result.status().message()
<< std::endl;
return 1;
}
auto& client = *client_result;
std::string channel = "cpp-events.consumer-group";
std::string group = "cpp-events-worker-group";
// Subscribe with a consumer group for load-balanced delivery.
std::cout << "[2] Subscribing with group=" << group << std::endl;
auto sub_result = client->SubscribeToEvents(
channel, group,
[](const kubemq::EventReceive& event) {
std::cout << "[4] Consumer group received: channel=" << event.channel
<< " body=" << event.body << std::endl;
},
[](const kubemq::Status& err) {
std::cerr << "[ERROR] Subscription error: " << err.message() << std::endl;
});
if (!sub_result.ok()) {
std::cerr << "[ERROR] SubscribeToEvents: " << sub_result.status().message() << std::endl;
return 1;
}
auto& sub = *sub_result;
std::this_thread::sleep_for(std::chrono::seconds(1));
// Publish an event to the group channel.
auto event_result = kubemq::Event::Builder()
.SetChannel(channel)
.SetBody("hello consumer group")
.SetMetadata("group-demo")
.Build();
if (!event_result.ok()) {
std::cerr << "[ERROR] Build event: " << event_result.status().message() << std::endl;
return 1;
}
auto send_status = client->SendEvent(*event_result);
if (!send_status.ok()) {
std::cerr << "[ERROR] SendEvent: " << send_status.message() << std::endl;
return 1;
}
std::cout << "[3] Event published to consumer group channel" << std::endl;
std::this_thread::sleep_for(std::chrono::seconds(2));
std::cout << "[5] Consumer group demo complete" << std::endl;
// Cancel the subscription explicitly.
// Note: Subscription destructor would also handle cleanup (RAII)
sub->Cancel();
auto close_status = client->Close();
if (!close_status.ok()) {
std::cerr << "[ERROR] Close failed: " << close_status.message() << std::endl;
return 1;
}
std::cout << "[6] Client closed" << std::endl;
return 0;
}How It Works
- Subscribes with a consumer group name (second parameter to
SubscribeToEvents()). - When multiple subscribers share the same group, each event is delivered to exactly one subscriber.
- Without a group (empty string), all subscribers receive every event (fan-out).
- Consumer groups enable horizontal scaling of event processors.
Related
Was this page helpful?