Redis RS 异步多路复用连接

示例说明

MultiplexedConnection 是 redis-rs 的甜点位：单条 TCP 连接被多个任务共享——命令内部排队管线化，并发调用方零连接成本。get_connection 包上 timeout 防止 Redis 无响应时启动挂死。cache-aside 模式里读缓存与查剩余寿命合并为一条管道命令，网络往返减半。

技术要点

• 多路复用让单连接服务并发任务
• 建连超时防启动挂死
• 管道合并读缓存与 TTL 查询

代码解读

warm_client 用两秒超时包裹连接获取，超时映射为 IoError 语义；cache_aside 的管道一次拿回值与 TTL，命中直接返回并报告剩余寿命，未命中执行加载器后 SET 带三百秒过期写回。KEYS 仅演示键枚举——生产环境务必换 SCAN 扫描避免阻塞。

使用建议

订阅类命令需要专用连接，多路复用连接不能跑 pubsub 阻塞命令。重试与断线重连用 connection manager 变体。键枚举一律 SCAN。

代码示例

use redis::{aio::MultiplexedConnection, Client, RedisResult, cmd};
use std::time::Duration;

async fn warm_client(url: &str) -> RedisResult<MultiplexedConnection> {
    let client = Client::open(url)?;

    let connection = tokio::time::timeout(
        Duration::from_secs(2),
        client.get_multiplexed_connection(),
    )
    .await
    .map_err(|_| redis::RedisError::from((redis::ErrorKind::IoError, "connect timeout")))??;

    Ok(connection)
}

async fn cache_aside(
    con: &mut MultiplexedConnection,
    key: &str,
    loader: impl FnOnce() -> String,
) -> RedisResult<String> {
    // pipeline 一并发出 GET 与 TTL，单次往返
    let (value, ttl): (Option<String>, i64) = redis::pipe()
        .cmd("GET")
        .arg(key)
        .ignore()
        .cmd("TTL")
        .arg(key)
        .query_async(con)
        .await?;

    if let Some(hit) = value {
        println!("{key} cached, ttl {ttl}s");
        return Ok(hit);
    }

    let fresh = loader();
    let _: () = cmd("SET")
        .arg(key)
        .arg(&fresh)
        .arg("EX")
        .arg(300)
        .query_async(con)
        .await?;
    println!("{key} loaded from source, cached 300s");
    Ok(fresh)
}

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

    let value = cache_aside(&mut con, "rates:usd", || "1.0".to_string()).await?;
    println!("rate: {value}");

    let pattern = "rates:*";
    let keys: Vec<String> = cmd("KEYS").arg(pattern).query_async(&mut con).await?;
    println!("cached keys: {keys:?}");

    let _: () = cmd("DEL").arg("rates:usd").query_async(&mut con).await?;
    Ok(())
}