Rust and Redis With Redis rs and Connection Managers

Redis commands are small and fast, so the cost of a request is mostly the network round trip. The client design follows: one multiplexed connection, shared. Everything compiles on Rust 1.90 with edition 2024, redis 0.27 (features tokio-comp, connection-manager) and tokio 1.

Connecting

use redis::{aio::ConnectionManager, AsyncCommands, Client};

#[tokio::main]
async fn main() -> redis::RedisResult<()> {
    let client = Client::open("redis://127.0.0.1:6379/")?;
    let mut con: ConnectionManager = client.get_connection_manager().await?;

    let _: () = con.set_ex("greeting", "hello", 60).await?;
    let value: Option<String> = con.get("greeting").await?;
    let missing: Option<String> = con.get("nope").await?;
    println!("{value:?} {missing:?}");

    let n: i64 = con.incr("hits", 1).await?;
    let removed: u64 = con.del("greeting").await?;
    println!("{n} {removed}");
    Ok(())
}

Some("hello") None
1 1

The let _: () = form tells the compiler to ignore the reply (OK). Return types drive conversion: ask for Option<String> when absence is normal, or String when it is an error. The ConnectionManager is Clone; give each task its own clone.

Sharing it

use redis::aio::ConnectionManager;

#[derive(Clone)]
struct AppState {
    redis: ConnectionManager,
}

async fn cached_len(state: &AppState, key: &str) -> redis::RedisResult<usize> {
    use redis::AsyncCommands;
    let mut con = state.redis.clone();
    let v: Option<String> = con.get(key).await?;
    Ok(v.map(|s| s.len()).unwrap_or(0))
}

All clones feed one connection, and Redis processes commands in order, so concurrency does not need more sockets. If the connection drops, the manager reconnects with backoff; commands issued meanwhile fail, so callers should treat Redis errors as recoverable (in-process caches cover the fallback).

Pipelines

use redis::{aio::ConnectionManager, RedisResult};

async fn bump_many(con: &mut ConnectionManager) -> RedisResult<(i64, i64)> {
    redis::pipe()
        .atomic()
        .incr("a", 1)
        .incr("b", 10)
        .query_async(con)
        .await
}

A pipeline sends the commands together and reads all replies, saving round trips. .atomic() wraps them in MULTI/EXEC so they apply as a unit.

Expiry and cache keys

Every cache key needs a TTL, or the dataset grows forever. Use set_ex or expire. For cache stampedes (many requests recomputing a missing key), take a short lock with SET key value NX EX 10 before recomputing. Key naming, invalidation and TTL choice are covered in cache invalidation.

Pub/sub

use futures::StreamExt;

async fn listen() -> redis::RedisResult<()> {
    let client = redis::Client::open("redis://127.0.0.1/")?;
    let mut pubsub = client.get_async_pubsub().await?;
    pubsub.subscribe("news").await?;
    let mut stream = pubsub.on_message();
    while let Some(msg) = stream.next().await {
        let payload: String = msg.get_payload()?;
        println!("{payload}");
    }
    Ok(())
}

A subscribed connection can only receive, so it is separate from the manager used for commands. Pub/sub messages are not stored: a subscriber that is offline misses them (use streams for durability).

Scripts and limits

Use redis::Script for atomic multi-step logic (rate limiters, compare-and-delete locks). Keep values small; very large values block the single-threaded server for everyone. The Redis pairing for limits is shown in rate limiting.