tdm_server_rust/cache/
redis_cache.rs

1//! Redis 缓存与协作状态后端
2//!
3//! ## 职责
4//!
5//! - 作为列表/统计类 JSON 缓存与结构化缓存的唯一后端，跨实例共享、带 TTL
6//! - 承载实时协作的页面锁、页面快照、跨实例广播（见编辑器协作模块）
7//!
8//! ## 启动约束
9//!
10//! 进程启动时必须能够连通 Redis，否则 [`RedisCache::connect`] 返回错误并终止启动。
11//!
12//! ## 键命名
13//!
14//! 所有键统一加 `{key_prefix}:` 前缀（默认 `tdm`）：
15//!
16//! | 逻辑键 | 实际键 | 用途 |
17//! |--------|--------|------|
18//! | `cache:json:<key>` | `tdm:cache:json:<key>` | 列表/统计 JSON 缓存 |
19//! | `cache:value:<namespace>:<key>` | `tdm:cache:value:<namespace>:<key>` | 结构化业务值缓存 |
20//! | `hub:{episodeId}:snapshot:<pageId>` | `tdm:hub:...` | 协作页面快照 |
21//! | `cache:invalidate`（频道） | `tdm:cache:invalidate` | 失效广播频道 |
22
23use crate::config::RedisConfig;
24use redis::aio::ConnectionManager;
25use redis::AsyncCommands;
26use serde::{de::DeserializeOwned, Serialize};
27use std::sync::Arc;
28
29/// 缓存失效广播频道（逻辑名，实际加前缀）
30pub const INVALIDATE_CHANNEL: &str = "cache:invalidate";
31
32/// Redis 缓存客户端
33///
34/// 克隆廉价：内部通过 [`Arc`] 与多路复用连接管理器共享同一连接。
35#[derive(Clone)]
36pub struct RedisCache {
37    /// 原始客户端（用于创建 pub/sub 专用连接）
38    client: redis::Client,
39    /// 多路复用异步连接管理器（自动重连）
40    conn: ConnectionManager,
41    /// 键统一前缀
42    key_prefix: Arc<String>,
43    /// JSON 缓存默认 TTL 秒数
44    default_ttl_secs: u64,
45    /// 本实例唯一发送者 ID（失效广播去重，避免自激重复 bump）
46    sender_id: Arc<String>,
47}
48
49impl RedisCache {
50    /// 连接 Redis 并校验连通性。
51    ///
52    /// ## 参数
53    /// - `cfg`: Redis 配置（URL / 前缀 / 默认 TTL）
54    ///
55    /// ## 返回
56    /// - `Ok(RedisCache)`: 连接成功且 PING 通过
57    /// - `Err`: 无法连接或 PING 失败（调用方应据此终止启动）
58    ///
59    /// ## 副作用
60    /// 建立到 Redis 的多路复用连接。
61    pub async fn connect(cfg: &RedisConfig) -> anyhow::Result<Self> {
62        let client = redis::Client::open(cfg.url.as_str())
63            .map_err(|e| anyhow::anyhow!("打开 Redis 客户端失败({}): {e}", cfg.url))?;
64        let mut conn = client
65            .get_connection_manager()
66            .await
67            .map_err(|e| anyhow::anyhow!("连接 Redis 失败({}): {e}", cfg.url))?;
68        // 启动期主动 PING 一次，确保 Redis 真正可用
69        redis::cmd("PING")
70            .query_async::<String>(&mut conn)
71            .await
72            .map_err(|e| anyhow::anyhow!("Redis PING 失败({}): {e}", cfg.url))?;
73        Ok(Self {
74            client,
75            conn,
76            key_prefix: Arc::new(cfg.key_prefix.clone()),
77            default_ttl_secs: cfg.default_ttl_secs,
78            sender_id: Arc::new(uuid::Uuid::new_v4().to_string()),
79        })
80    }
81
82    /// 本实例唯一发送者 ID
83    pub fn sender_id(&self) -> &str {
84        &self.sender_id
85    }
86
87    /// JSON 缓存默认 TTL 秒数
88    pub fn default_ttl_secs(&self) -> u64 {
89        self.default_ttl_secs
90    }
91
92    /// 把逻辑键加上统一前缀，得到 Redis 实际键
93    pub fn full_key(&self, logical: &str) -> String {
94        format!("{}:{}", self.key_prefix, logical)
95    }
96
97    /// 生成 JSON 响应缓存逻辑键。
98    pub fn json_cache_key(key: &str) -> String {
99        format!("cache:json:{key}")
100    }
101
102    /// 生成结构化值缓存逻辑键。
103    pub fn value_cache_key(namespace: &str, key: &str) -> String {
104        format!("cache:value:{namespace}:{key}")
105    }
106
107    /// 读取字节值。
108    ///
109    /// 运行期 Redis 抖动时记录告警并返回 `None`（读路径降级，不影响请求）。
110    pub async fn get_bytes(&self, logical: &str) -> Option<Vec<u8>> {
111        let key = self.full_key(logical);
112        let mut conn = self.conn.clone();
113        match conn.get::<_, Option<Vec<u8>>>(&key).await {
114            Ok(v) => v,
115            Err(e) => {
116                tracing::warn!("Redis GET 失败 key={key}: {e}");
117                None
118            }
119        }
120    }
121
122    /// 写入字节值并设置 TTL（秒）。
123    ///
124    /// 运行期失败仅记录告警，不向上抛错（写缓存属尽力而为）。
125    pub async fn set_bytes_ttl(&self, logical: &str, value: &[u8], ttl_secs: u64) {
126        let key = self.full_key(logical);
127        let mut conn = self.conn.clone();
128        if let Err(e) = conn.set_ex::<_, _, ()>(&key, value, ttl_secs).await {
129            tracing::warn!("Redis SETEX 失败 key={key}: {e}");
130        }
131    }
132
133    /// 删除键。
134    pub async fn del(&self, logical: &str) {
135        let key = self.full_key(logical);
136        let mut conn = self.conn.clone();
137        if let Err(e) = conn.del::<_, ()>(&key).await {
138            tracing::warn!("Redis DEL 失败 key={key}: {e}");
139        }
140    }
141
142    /// 按逻辑键模式删除 Redis 键。
143    ///
144    /// 使用 `SCAN MATCH` 分批删除，避免 `KEYS` 阻塞 Redis。
145    pub async fn del_pattern(&self, logical_pattern: &str) {
146        let pattern = self.full_key(logical_pattern);
147        let mut conn = self.conn.clone();
148        let mut cursor = 0_u64;
149        loop {
150            let scan = redis::cmd("SCAN")
151                .arg(cursor)
152                .arg("MATCH")
153                .arg(&pattern)
154                .arg("COUNT")
155                .arg(100_u32)
156                .query_async::<(u64, Vec<String>)>(&mut conn)
157                .await;
158            let (next, keys) = match scan {
159                Ok(v) => v,
160                Err(e) => {
161                    tracing::warn!("Redis SCAN 失败 pattern={pattern}: {e}");
162                    return;
163                }
164            };
165            if !keys.is_empty() {
166                if let Err(e) = redis::cmd("DEL")
167                    .arg(keys.as_slice())
168                    .query_async::<()>(&mut conn)
169                    .await
170                {
171                    tracing::warn!("Redis DEL 批量失败 pattern={pattern}: {e}");
172                }
173            }
174            if next == 0 {
175                break;
176            }
177            cursor = next;
178        }
179    }
180
181    /// 读取预序列化 JSON 响应缓存。
182    pub async fn get_json_bytes(&self, key: &str) -> Option<Arc<Vec<u8>>> {
183        self.get_bytes(&Self::json_cache_key(key))
184            .await
185            .map(Arc::new)
186    }
187
188    /// 写入预序列化 JSON 响应缓存。
189    pub async fn set_json_bytes_ttl(&self, key: &str, value: &[u8], ttl_secs: u64) {
190        self.set_bytes_ttl(&Self::json_cache_key(key), value, ttl_secs)
191            .await;
192    }
193
194    /// 删除单个 JSON 响应缓存。
195    pub async fn del_json(&self, key: &str) {
196        self.del(&Self::json_cache_key(key)).await;
197    }
198
199    /// 按业务键模式删除 JSON 响应缓存。
200    pub async fn del_json_pattern(&self, pattern: &str) {
201        self.del_pattern(&Self::json_cache_key(pattern)).await;
202    }
203
204    /// 读取结构化业务值缓存。
205    pub async fn get_value<T>(&self, namespace: &str, key: &str) -> Option<T>
206    where
207        T: DeserializeOwned,
208    {
209        let logical = Self::value_cache_key(namespace, key);
210        let bytes = self.get_bytes(&logical).await?;
211        match serde_json::from_slice::<T>(&bytes) {
212            Ok(v) => Some(v),
213            Err(e) => {
214                tracing::warn!("Redis JSON 反序列化失败 key={logical}: {e}");
215                None
216            }
217        }
218    }
219
220    /// 写入结构化业务值缓存。
221    pub async fn set_value_ttl<T>(&self, namespace: &str, key: &str, value: &T, ttl_secs: u64)
222    where
223        T: Serialize,
224    {
225        let logical = Self::value_cache_key(namespace, key);
226        let bytes = match serde_json::to_vec(value) {
227            Ok(v) => v,
228            Err(e) => {
229                tracing::warn!("Redis JSON 序列化失败 key={logical}: {e}");
230                return;
231            }
232        };
233        self.set_bytes_ttl(&logical, &bytes, ttl_secs).await;
234    }
235
236    /// 删除单个结构化业务值缓存。
237    pub async fn del_value(&self, namespace: &str, key: &str) {
238        self.del(&Self::value_cache_key(namespace, key)).await;
239    }
240
241    /// 按业务键模式删除结构化业务值缓存。
242    pub async fn del_value_pattern(&self, namespace: &str, pattern: &str) {
243        self.del_pattern(&Self::value_cache_key(namespace, pattern))
244            .await;
245    }
246
247    /// 写入字符串值并设置 TTL（秒）。
248    pub async fn set_str_ttl(&self, logical: &str, value: &str, ttl_secs: u64) {
249        self.set_bytes_ttl(logical, value.as_bytes(), ttl_secs)
250            .await;
251    }
252
253    /// 读取字符串值。
254    pub async fn get_str(&self, logical: &str) -> Option<String> {
255        self.get_bytes(logical)
256            .await
257            .and_then(|b| String::from_utf8(b).ok())
258    }
259
260    /// 条件写入：仅当键不存在时写入并设置 TTL（`SET key val NX EX ttl`）。
261    ///
262    /// 用于协作页面锁的原子抢占。
263    ///
264    /// ## 返回
265    /// - `true`：本次成功抢占（键原先不存在）
266    /// - `false`：键已存在（被他人持有）或 Redis 异常
267    pub async fn set_nx_ex(&self, logical: &str, value: &str, ttl_secs: u64) -> bool {
268        let key = self.full_key(logical);
269        let mut conn = self.conn.clone();
270        match redis::cmd("SET")
271            .arg(&key)
272            .arg(value)
273            .arg("NX")
274            .arg("EX")
275            .arg(ttl_secs)
276            .query_async::<Option<String>>(&mut conn)
277            .await
278        {
279            Ok(v) => v.is_some(),
280            Err(e) => {
281                tracing::warn!("Redis SET NX 失败 key={key}: {e}");
282                false
283            }
284        }
285    }
286
287    /// 刷新键 TTL（秒）。用于续约页面锁。
288    pub async fn expire(&self, logical: &str, ttl_secs: u64) {
289        let key = self.full_key(logical);
290        let mut conn = self.conn.clone();
291        if let Err(e) = conn.expire::<_, ()>(&key, ttl_secs as i64).await {
292            tracing::warn!("Redis EXPIRE 失败 key={key}: {e}");
293        }
294    }
295
296    /// 写入哈希字段并刷新整哈希 TTL（用于项目在线状态表）。
297    pub async fn hset_ttl(&self, logical: &str, field: &str, value: &str, ttl_secs: u64) {
298        let key = self.full_key(logical);
299        let mut conn = self.conn.clone();
300        if let Err(e) = conn.hset::<_, _, _, ()>(&key, field, value).await {
301            tracing::warn!("Redis HSET 失败 key={key}: {e}");
302            return;
303        }
304        if let Err(e) = conn.expire::<_, ()>(&key, ttl_secs as i64).await {
305            tracing::warn!("Redis EXPIRE 失败 key={key}: {e}");
306        }
307    }
308
309    /// 读取哈希全部字段（field, value）。
310    pub async fn hgetall(&self, logical: &str) -> Vec<(String, String)> {
311        let key = self.full_key(logical);
312        let mut conn = self.conn.clone();
313        match conn.hgetall::<_, Vec<(String, String)>>(&key).await {
314            Ok(v) => v,
315            Err(e) => {
316                tracing::warn!("Redis HGETALL 失败 key={key}: {e}");
317                Vec::new()
318            }
319        }
320    }
321
322    /// 删除哈希字段。
323    pub async fn hdel(&self, logical: &str, field: &str) {
324        let key = self.full_key(logical);
325        let mut conn = self.conn.clone();
326        if let Err(e) = conn.hdel::<_, _, ()>(&key, field).await {
327            tracing::warn!("Redis HDEL 失败 key={key}: {e}");
328        }
329    }
330
331    /// 发布一条消息到指定逻辑频道。
332    pub async fn publish(&self, channel: &str, payload: &str) {
333        let ch = self.full_key(channel);
334        let mut conn = self.conn.clone();
335        if let Err(e) = conn.publish::<_, _, ()>(&ch, payload).await {
336            tracing::warn!("Redis PUBLISH 失败 channel={ch}: {e}");
337        }
338    }
339
340    /// 订阅指定逻辑频道，对每条消息调用 `handler`。
341    ///
342    /// 内部使用独立 pub/sub 连接；连接异常时按退避重连，持续运行直到进程退出。
343    ///
344    /// ## 参数
345    /// - `channel`: 逻辑频道名（自动加前缀）
346    /// - `handler`: 收到消息载荷时的回调
347    pub async fn subscribe_loop<F>(self, channel: &str, mut handler: F)
348    where
349        F: FnMut(String) + Send + 'static,
350    {
351        let ch = self.full_key(channel);
352        loop {
353            match self.client.get_async_pubsub().await {
354                Ok(mut pubsub) => {
355                    if let Err(e) = pubsub.subscribe(&ch).await {
356                        tracing::warn!("Redis SUBSCRIBE 失败 channel={ch}: {e}");
357                        tokio::time::sleep(std::time::Duration::from_secs(2)).await;
358                        continue;
359                    }
360                    tracing::info!("Redis 订阅频道 {ch} 成功");
361                    use futures::StreamExt;
362                    let mut stream = pubsub.on_message();
363                    while let Some(msg) = stream.next().await {
364                        if let Ok(payload) = msg.get_payload::<String>() {
365                            handler(payload);
366                        }
367                    }
368                    tracing::warn!("Redis 订阅流中断 channel={ch}，准备重连");
369                }
370                Err(e) => {
371                    tracing::warn!("Redis 创建 pubsub 连接失败 channel={ch}: {e}");
372                }
373            }
374            tokio::time::sleep(std::time::Duration::from_secs(2)).await;
375        }
376    }
377}
tdm_server_rust/cache/redis_cache.rs

tdm_server_rust/cache/
redis_cache.rs
