Query Group
Load-balance KubeMQ Queries across handlers with consumer groups so each query reaches one handler, in Rust.
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 call: client.subscribe_to_queries(channel, group, ...) with a non-empty group 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 - Rust SDK installed (
cargo add kubemq)
Code
use kubemq::prelude::*;
use kubemq::{QueryBuilder, QueryReplyBuilder};
use std::time::Duration;
#[tokio::main]
async fn main() -> kubemq::Result<()> {
let client = KubemqClient::builder()
.host("localhost")
.port(50000)
.build()
.await?;
let channel = "rust-rpc.query-group";
let group = "query-handler-group";
let rc1 = client.clone();
let sub1 = client.subscribe_to_queries(channel, group,
move |q| { let c = rc1.clone(); Box::pin(async move {
println!("Handler-1: id={}", q.id);
let reply = QueryReplyBuilder::new().request_id(&q.id).response_to(&q.response_to)
.body(b"from-handler-1".to_vec()).build();
let _ = c.send_query_response(reply).await;
})}, None,
).await?;
let rc2 = client.clone();
let sub2 = client.subscribe_to_queries(channel, group,
move |q| { let c = rc2.clone(); Box::pin(async move {
println!("Handler-2: id={}", q.id);
let reply = QueryReplyBuilder::new().request_id(&q.id).response_to(&q.response_to)
.body(b"from-handler-2".to_vec()).build();
let _ = c.send_query_response(reply).await;
})}, None,
).await?;
tokio::time::sleep(Duration::from_millis(500)).await;
for i in 0..5 {
let query = QueryBuilder::new()
.channel(channel)
.body(format!("query-{}", i).into_bytes())
.timeout(Duration::from_secs(10))
.build();
let resp = client.send_query(query).await?;
println!("Response {}: body={}", i, String::from_utf8_lossy(&resp.body));
}
sub1.unsubscribe().await;
sub2.unsubscribe().await;
client.close().await?;
Ok(())
}How It Works
- Both handlers join the same
group, so each query is delivered to exactly one handler. - The server distributes queries round-robin across group members.
- Review timeouts, channel names, and client IDs before running against shared environments.
- Run the program while the server from the prerequisites is available.
Related
Was this page helpful?