consistent_hash

consistent_hash spreads a stream over queues bound with integer weights, by a hash of the routing key, through the RabbitMQ plugin.

The constructor exists with the crate's plugin-consistent-hash feature and names the x-consistent-hash type of the rabbitmq_consistent_hash_exchange plugin. The server needs the plugin enabled: rabbitmq-plugins enable rabbitmq_consistent_hash_exchange. Each queue binds with its weight as the routing key, such as "1" or "2", and takes that share of the hash space. The exchange hashes the routing key of each message, so messages with one key always reach the same queue. So a key such as a customer or an account keeps its messages in one queue, in order, while the stream spreads over all of them. Within one service instance, lanes give the same per-key order without the plugin.

The plugin does the routing on the server. The in-process transport declares the exchange and its bindings and routes nothing through it, so a test of the spread runs against a plugin-enabled server with TestApp::start_live.

Example

Orders are hashed by customer over two shards of equal weight.

Macros

examples/lapin_exchanges_consistent_hash.rs

/// Two shards of equal weight: the binding key is the weight, and one customer's orders always
/// land on the same shard.
#[subscriber(RabbitQueue::new("orders.shard-a")
    .bind(RabbitExchange::consistent_hash("orders-by-customer"), "1"))]
async fn shard_a(order: &Order) -> HandlerOutcome {
    tracing::info!(order.id, order.customer, "shard a");
    HandlerOutcome::ack()
}

#[subscriber(RabbitQueue::new("orders.shard-b")
    .bind(RabbitExchange::consistent_hash("orders-by-customer"), "1"))]
async fn shard_b(order: &Order) -> HandlerOutcome {
    tracing::info!(order.id, order.customer, "shard b");
    HandlerOutcome::ack()
}

Manual

examples/manual/lapin_exchanges_consistent_hash.rs

struct ShardA;

impl Handle<Order> for ShardA {
    fn handle(
        &self,
        order: &Order,
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<(), HandlerOutcome>> {
        tracing::info!(order.id, order.customer, "shard a");
        ready(Ok(()))
    }
}

struct ShardB;

impl Handle<Order> for ShardB {
    fn handle(
        &self,
        order: &Order,
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<(), HandlerOutcome>> {
        tracing::info!(order.id, order.customer, "shard b");
        ready(Ok(()))
    }
}

/// Two shards of equal weight: the binding key is the weight, and one customer's orders always
/// land on the same shard.
fn shards() -> (RabbitQueue, RabbitQueue) {
    let by_customer = RabbitExchange::consistent_hash("orders-by-customer");
    let a = RabbitQueue::new("orders.shard-a").bind(by_customer.clone(), "1");
    let b = RabbitQueue::new("orders.shard-b").bind(by_customer, "1");
    (a, b)
}