Redis 发布订阅通道消费

示例说明

pubsub 的订阅命令会阻塞连接，不能与普通命令共用——get_async_pubsub 返回专用连接是正确姿势。on_message 返回 future，配合 timeout_at 让订阅循环拥有截止时间：既能在无消息时优雅退出，也让长期监听有健康检查的空隙。发布端用多路复用连接即可，双向各司其职。

技术要点

• 订阅走专用 pubsub 连接
• timeout_at 给循环加截止时间
• 按通道名区分消息来源

代码解读

listen 先订阅全部目标通道再进入消息循环，每条消息提取通道名与载荷拼接记录；deadline 到点或收满即退出。main 里监听任务先启动，等一百毫秒确保订阅生效后连续发布两条不同通道消息——异步竞态是 pubsub 最常见的坑，订阅确认前的发布会丢失。

使用建议

pubsub 不持久化，离线消息会丢——可靠投递改用 Streams。断线重连要重新订阅全部通道。生产监听器配合心跳与慢消费者监控。

代码示例

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

async fn listen(
    url: &str,
    channels: &[&str],
    max_messages: usize,
) -> RedisResult<Vec<String>> {
    let client = Client::open(url)?;
    let mut received = Vec::new();

    let mut pubsub = client.get_async_pubsub().await?;
    for channel in channels {
        pubsub.subscribe(channel).await?;
    }

    // 收满或超时即退出：演示可中断的订阅循环
    let deadline = tokio::time::Instant::now() + Duration::from_secs(5);
    while received.len() < max_messages {
        let next = tokio::time::timeout_at(deadline, pubsub.on_message()).await;
        match next {
            Ok(Ok(msg)) => {
                let channel = msg.get_channel_name();
                let payload: String = msg.get_payload()?;
                received.push(format!("{channel}: {payload}"));
            }
            _ => break, // 超时或错误：跳出循环
        }
    }
    Ok(received)
}

#[tokio::main]
async fn main() -> RedisResult<()> {
    let url = "redis://127.0.0.1:6379";

    // 发布端：独立连接直接 PUBLISH
    let publisher = Client::open(url)?;
    let mut pub_con = publisher.get_multiplexed_connection().await?;

    let listener = tokio::spawn(listen(url, &["events:billing", "events:auth"], 3));

    tokio::time::sleep(Duration::from_millis(100)).await;
    let _: () = redis::cmd("PUBLISH")
        .arg("events:billing")
        .arg("invoice-settled")
        .query_async(&mut pub_con)
        .await?;
    let _: () = redis::cmd("PUBLISH")
        .arg("events:auth")
        .arg("user-login")
        .query_async(&mut pub_con)
        .await?;

    let messages = listener.await.unwrap()?;
    for msg in &messages {
        println!("{msg}");
    }
    Ok(())
}