mycommon_utils/database/
redis_util.rs

1use std::sync::Arc;
2use redis::{FromRedisValue, RedisResult, ToRedisArgs};
3use crate::database::config::{ REDIS_ASYNC_POOL};
4
5/// 同步 Redis 工具类，连接在实例 drop 时释放
6// pub struct RedisUtil {
7//     connection: PooledConnection<Client>
8// }
9
10// impl RedisUtil {
11//     pub fn new() -> Result<RedisUtil> {
12//         let client = REDIS_POOL.get().ok_or_else(|| {
13//             tracing::error!("Failed to get Redis connection pool");
14//             Error::NotFound()
15//         })?;
16
17//         let connection = client.get().map_err(|e| {
18//             tracing::error!("Failed to obtain Redis connection: {:?}", e);
19//             Error::NotFound()
20//         })?;
21
22//         Ok(RedisUtil { connection })
23//     }
24
25//     pub fn set<T: ToRedisArgs>(&mut self, key: &str, value: &T) -> RedisResult<bool> {
26//         let result = self.connection.set(key, value)?;
27//         Ok(result)
28//     }
29
30//     pub fn set_with_expiry<T: ToRedisArgs>(&mut self, key: &str, value: &T, expiry_seconds: u64) -> RedisResult<()> {
31//         let result = self.connection.set_ex(key, value, expiry_seconds)?;
32//         Ok(result)
33//     }
34
35//     pub fn set_nx_ex<T: ToRedisArgs>(&mut self, key: &str, value: &T, expiry_seconds: u64) -> RedisResult<bool> {
36//         let mut cmd = redis::Cmd::set(key, value);
37//         cmd.arg("NX").arg("EX").arg(expiry_seconds);
38//         let result: String = cmd.query(&mut self.connection)?;
39//         Ok(result == "OK")
40//     }
41
42//     pub fn expire(&mut self, key: &str, expiry_seconds: i64) -> RedisResult<bool> {
43//         let result = self.connection.expire(key, expiry_seconds)?;
44//         Ok(result)
45//     }
46
47//     pub fn get<T: FromRedisValue>(&mut self, key: &str) -> RedisResult<Option<T>> {
48//         let value: Option<T> = self.connection.get(key)?;
49//         Ok(value)
50//     }
51
52//     pub fn pipeline_operations<T: ToRedisArgs>(&mut self, keys: Vec<&str>, values: Vec<&T>) -> RedisResult<()> {
53//         if keys.len() != values.len() {
54//             return Err(redis::RedisError::from((
55//                 redis::ErrorKind::TypeError,
56//                 "Keys and values must have the same length",
57//             )));
58//         }
59//         let mut pipe = redis::pipe();
60//         for (key, value) in keys.iter().zip(values.iter()) {
61//             pipe.set(*key, *value).ignore();
62//         }
63//         pipe.query::<()>(&mut self.connection)?;
64//         Ok(())
65//     }
66
67//     pub fn exists(&mut self, key: &str) -> RedisResult<bool> {
68//         let exists = self.connection.exists(key)?;
69//         Ok(exists)
70//     }
71
72//     pub fn del_vec(&mut self, keys: Vec<&str>) -> RedisResult<usize> {
73//         self.connection.del(keys)
74//     }
75
76//     pub fn del(&mut self, key: &str) -> RedisResult<usize> {
77//         self.connection.del(key)
78//     }
79
80//     pub fn ttl(&mut self, key: &str) -> RedisResult<isize> {
81//         self.connection.ttl(key)
82//     }
83
84//     pub fn incr(&mut self, key: &str) -> RedisResult<isize> {
85//         self.connection.incr(key, 1)
86//     }
87
88//     pub fn decr(&mut self, key: &str) -> RedisResult<isize> {
89//         self.connection.decr(key, 1)
90//     }
91
92//     pub fn hset(&mut self, key: &str, field: &str, value: &str) -> RedisResult<()> {
93//         self.connection.hset(key, field, value)
94//     }
95
96//     pub fn hget(&mut self, key: &str, field: &str) -> RedisResult<Option<String>> {
97//         self.connection.hget(key, field)
98//     }
99
100//     pub fn lpush<T: FromRedisValue>(&mut self, key: &str, value: &str) -> RedisResult<T> {
101//         self.connection.lpush(key, value)
102//     }
103
104//     pub fn rpop<T: ToString>(&mut self, key: &str, count: Option<usize>) -> RedisResult<Option<Vec<String>>> {
105//         use std::num::NonZeroUsize;
106//         let non_zero_count = count.and_then(NonZeroUsize::new);
107//         self.connection.rpop(key, non_zero_count)
108//     }
109// }
110
111/// 异步 Redis 工具类，适用于 Tokio 异步运行时
112/// 推荐在 Axum 中间件和异步代码中使用
113/// 每个方法独立获取和释放连接，避免连接长时间占用
114pub struct RedisAsync;
115
116impl RedisAsync {
117    fn pool() -> &'static Arc<deadpool_redis::Pool> {
118        REDIS_ASYNC_POOL.get().expect("Redis async pool not initialized")
119    }
120
121    /// 异步设置键值对
122    pub async fn set<T: ToRedisArgs + Send + Sync>(key: &str, value: &T) -> RedisResult<()> {
123        let mut conn = Self::pool().get().await.map_err(|e| {
124            redis::RedisError::from((redis::ErrorKind::Io, "Pool error", format!("{:?}", e)))
125        })?;
126        redis::cmd("SET").arg(key).arg(value).query_async(&mut *conn).await
127    }
128
129    /// 异步设置带过期时间的键值对
130    pub async fn set_with_expiry<T: ToRedisArgs + Send + Sync>(key: &str, value: &T, expiry_seconds: u64) -> RedisResult<()> {
131        let mut conn = Self::pool().get().await.map_err(|e| {
132            redis::RedisError::from((redis::ErrorKind::Io, "Pool error", format!("{:?}", e)))
133        })?;
134        redis::cmd("SET")
135            .arg(key)
136            .arg(value)
137            .arg("EX")
138            .arg(expiry_seconds)
139            .query_async(&mut *conn).await
140    }
141
142    /// 异步设置带 NX 和 EX 的键值对（分布式锁）
143    pub async fn set_nx_ex<T: ToRedisArgs + Send + Sync>(key: &str, value: &T, expiry_seconds: u64) -> RedisResult<bool> {
144        let mut conn = Self::pool().get().await.map_err(|e| {
145            redis::RedisError::from((redis::ErrorKind::Io, "Pool error", format!("{:?}", e)))
146        })?;
147        let result: Option<String> = redis::cmd("SET")
148            .arg(key)
149            .arg(value)
150            .arg("NX")
151            .arg("EX")
152            .arg(expiry_seconds)
153            .query_async(&mut *conn).await?;
154        Ok(result == Some("OK".to_string()))
155    }
156
157    /// 异步获取键对应的值
158    pub async fn get<T: FromRedisValue + Send>(key: &str) -> RedisResult<Option<T>> {
159        let mut conn = Self::pool().get().await.map_err(|e| {
160            redis::RedisError::from((redis::ErrorKind::Io, "Pool error", format!("{:?}", e)))
161        })?;
162        redis::cmd("GET").arg(key).query_async(&mut *conn).await
163    }
164
165    /// 异步设置过期时间
166    pub async fn expire(key: &str, expiry_seconds: i64) -> RedisResult<bool> {
167        let mut conn = Self::pool().get().await.map_err(|e| {
168            redis::RedisError::from((redis::ErrorKind::Io, "Pool error", format!("{:?}", e)))
169        })?;
170        let result: i32 = redis::cmd("EXPIRE").arg(key).arg(expiry_seconds).query_async(&mut *conn).await?;
171        Ok(result == 1)
172    }
173
174    /// 异步检查键是否存在
175    pub async fn exists(key: &str) -> RedisResult<bool> {
176        let mut conn = Self::pool().get().await.map_err(|e| {
177            redis::RedisError::from((redis::ErrorKind::Io, "Pool error", format!("{:?}", e)))
178        })?;
179        let exists: i32 = redis::cmd("EXISTS").arg(key).query_async(&mut *conn).await?;
180        Ok(exists == 1)
181    }
182
183    /// 异步删除键
184    pub async fn del(key: &str) -> RedisResult<i32> {
185        let mut conn = Self::pool().get().await.map_err(|e| {
186            redis::RedisError::from((redis::ErrorKind::Io, "Pool error", format!("{:?}", e)))
187        })?;
188        redis::cmd("DEL").arg(key).query_async(&mut *conn).await
189    }
190
191    /// 异步删除多个键
192    pub async fn del_vec(keys: &[&str]) -> RedisResult<i32> {
193        let mut conn = Self::pool().get().await.map_err(|e| {
194            redis::RedisError::from((redis::ErrorKind::Io, "Pool error", format!("{:?}", e)))
195        })?;
196        redis::cmd("DEL").arg(keys).query_async(&mut *conn).await
197    }
198
199    /// 异步获取 TTL
200    pub async fn ttl(key: &str) -> RedisResult<i64> {
201        let mut conn = Self::pool().get().await.map_err(|e| {
202            redis::RedisError::from((redis::ErrorKind::Io, "Pool error", format!("{:?}", e)))
203        })?;
204        redis::cmd("TTL").arg(key).query_async(&mut *conn).await
205    }
206
207    /// 异步自增
208    pub async fn incr(key: &str) -> RedisResult<i64> {
209        let mut conn = Self::pool().get().await.map_err(|e| {
210            redis::RedisError::from((redis::ErrorKind::Io, "Pool error", format!("{:?}", e)))
211        })?;
212        redis::cmd("INCR").arg(key).query_async(&mut *conn).await
213    }
214
215    /// 异步自减
216    pub async fn decr(key: &str) -> RedisResult<i64> {
217        let mut conn = Self::pool().get().await.map_err(|e| {
218            redis::RedisError::from((redis::ErrorKind::Io, "Pool error", format!("{:?}", e)))
219        })?;
220        redis::cmd("DECR").arg(key).query_async(&mut *conn).await
221    }
222
223    /// 异步设置哈希字段
224    pub async fn hset<K: ToRedisArgs + Send + Sync, V: ToRedisArgs + Send + Sync>(key: &str, field: K, value: V) -> RedisResult<()> {
225        let mut conn = Self::pool().get().await.map_err(|e| {
226            redis::RedisError::from((redis::ErrorKind::Io, "Pool error", format!("{:?}", e)))
227        })?;
228        redis::cmd("HSET").arg(key).arg(field).arg(value).query_async(&mut *conn).await
229    }
230
231    /// 异步获取哈希字段
232    pub async fn hget<T: FromRedisValue + Send>(key: &str, field: &str) -> RedisResult<Option<T>> {
233        let mut conn = Self::pool().get().await.map_err(|e| {
234            redis::RedisError::from((redis::ErrorKind::Io, "Pool error", format!("{:?}", e)))
235        })?;
236        redis::cmd("HGET").arg(key).arg(field).query_async(&mut *conn).await
237    }
238
239    /// 异步向列表左侧推入元素
240    pub async fn lpush<T: ToRedisArgs + Send + Sync>(key: &str, value: T) -> RedisResult<i32> {
241        let mut conn = Self::pool().get().await.map_err(|e| {
242            redis::RedisError::from((redis::ErrorKind::Io, "Pool error", format!("{:?}", e)))
243        })?;
244        redis::cmd("LPUSH").arg(key).arg(value).query_async(&mut *conn).await
245    }
246
247    /// 异步从列表右侧弹出元素
248    pub async fn rpop(key: &str) -> RedisResult<Option<String>> {
249        let mut conn = Self::pool().get().await.map_err(|e| {
250            redis::RedisError::from((redis::ErrorKind::Io, "Pool error", format!("{:?}", e)))
251        })?;
252        redis::cmd("RPOP").arg(key).query_async(&mut *conn).await
253    }
254
255    /// 异步添加 Stream 消息（带自动清理）
256    pub async fn xadd_maxlen<K: ToRedisArgs + Send + Sync, V: ToRedisArgs + Send + Sync>(
257        key: &str,
258        maxlen: usize,
259        field: K,
260        value: V,
261    ) -> RedisResult<String> {
262        let mut conn = Self::pool().get().await.map_err(|e| {
263            redis::RedisError::from((redis::ErrorKind::Io, "Pool error", format!("{:?}", e)))
264        })?;
265        redis::cmd("XADD")
266            .arg(key)
267            .arg("MAXLEN")
268            .arg("~")
269            .arg(maxlen)
270            .arg("*")
271            .arg(field)
272            .arg(value)
273            .query_async(&mut *conn)
274            .await
275    }
276    /// 异步删除 Stream 消息
277    pub async fn xdel<ID: ToRedisArgs + Send + Sync>(key: &str, ids: &[&ID]) -> RedisResult<i32> {
278        let mut conn = Self::pool().get().await.map_err(|e| {
279            redis::RedisError::from((redis::ErrorKind::Io, "Pool error", format!("{:?}", e)))
280        })?;
281        redis::cmd("XDEL").arg(key).arg(ids).query_async(&mut *conn).await
282    }
283}
mycommon_utils/database/redis_util.rs

mycommon_utils/database/
redis_util.rs
