/** * Redis 缓存模块 * 用于缓存聚合数据,减少数据库查询 */ const Redis = require('ioredis'); const { getRuntimeRedisConfig } = require('../services/redisConfig'); let redis = null; let connected = false; let currentConfigSignature = ''; let unavailableUntil = 0; let configCache = { expiresAt: 0, value: null, }; const REDIS_FAILURE_COOLDOWN_MS = Number.parseInt(process.env.REDIS_FAILURE_COOLDOWN_MS || '30000', 10); const REDIS_CONNECT_TIMEOUT_MS = Number.parseInt(process.env.REDIS_CONNECT_TIMEOUT_MS || '1000', 10); const REDIS_OPERATION_TIMEOUT_MS = Math.min( Math.max(Number.parseInt(process.env.REDIS_OPERATION_TIMEOUT_MS || '500', 10), 100), 2000 ); const REDIS_CONFIG_RESOLVE_TIMEOUT_MS = Math.min( Math.max(Number.parseInt(process.env.REDIS_CONFIG_RESOLVE_TIMEOUT_MS || '1500', 10), 300), 3000 ); function withTimeout(promise, timeoutMs, fallback) { let timer; const timeout = new Promise((resolve) => { timer = setTimeout(() => resolve(fallback), timeoutMs); }); return Promise.race([promise, timeout]).finally(() => { if (timer) { clearTimeout(timer); } }); } function buildRedisClientConfig(config) { return { host: config.host, port: parseInt(config.port || '6379', 10), username: config.username || process.env.REDIS_USERNAME || process.env.REDIS_USER || undefined, password: config.password || undefined, db: parseInt(config.db || '0', 10), tls: config.tls ? { servername: config.tlsServerName || config.host } : undefined, connectTimeout: REDIS_CONNECT_TIMEOUT_MS, retryStrategy: (times) => (times > 1 ? null : 200), maxRetriesPerRequest: 1, enableOfflineQueue: false, enableReadyCheck: false, lazyConnect: true, }; } function getConfigSignature(config) { if (!config?.host) { return ''; } return JSON.stringify({ host: config.host, port: parseInt(config.port || '6379', 10), db: parseInt(config.db || '0', 10), username: config.username || process.env.REDIS_USERNAME || process.env.REDIS_USER || '', password: config.password || '', tls: Boolean(config.tls), tlsServerName: config.tlsServerName || '', }); } async function resolveRedisConfig(force = false) { const now = Date.now(); if (!force && configCache.expiresAt > now) { return configCache.value; } const config = await getRuntimeRedisConfig(); configCache = { value: config, expiresAt: now + 5000, }; return config; } async function closeCurrentClient() { if (!redis) { connected = false; currentConfigSignature = ''; return; } const client = redis; redis = null; connected = false; currentConfigSignature = ''; try { await client.quit(); } catch (_error) { client.disconnect(); } } async function ensureRedisConnection(client) { if (!client) { return false; } if (unavailableUntil > Date.now()) { return false; } // `connect` only means that the socket is being opened. Sending a command // in that state can wait for ioredis' retry window, so only ready is usable. if (client.status === 'ready') { return true; } if (client.status === 'wait' || client.status === 'end') { try { await client.connect(); unavailableUntil = 0; return true; } catch (error) { unavailableUntil = Date.now() + REDIS_FAILURE_COOLDOWN_MS; connected = false; console.warn('[Redis] 连接失败,进入冷却期:', error.message); return false; } } return false; } /** * 获取 Redis 实例 */ const getRedis = async () => { const redisConfig = await withTimeout( resolveRedisConfig(), REDIS_CONFIG_RESOLVE_TIMEOUT_MS, null ); if (!redisConfig) { await closeCurrentClient(); return null; } const nextSignature = getConfigSignature(redisConfig); if (redis && currentConfigSignature !== nextSignature) { await closeCurrentClient(); } if (!redis) { redis = new Redis(buildRedisClientConfig(redisConfig)); currentConfigSignature = nextSignature; redis.on('connect', () => { connected = true; unavailableUntil = 0; console.log('[Redis] 连接成功'); }); redis.on('error', (err) => { connected = false; unavailableUntil = Date.now() + REDIS_FAILURE_COOLDOWN_MS; console.error('[Redis] 连接错误:', err.message); }); redis.on('close', () => { connected = false; unavailableUntil = Date.now() + REDIS_FAILURE_COOLDOWN_MS; console.log('[Redis] 连接关闭'); }); } return redis; }; /** * 安全获取缓存,Redis 不可用时返回 null */ const get = async (key) => { try { const r = await getRedis(); if (!r) { return null; } if (!(await ensureRedisConnection(r))) { return null; } const value = await withTimeout(r.get(key), REDIS_OPERATION_TIMEOUT_MS, null); return value ? JSON.parse(value) : null; } catch (error) { console.warn('[Redis] GET 失败:', error.message); return null; } }; /** * 原子读取并删除缓存。用于一次性票据,避免并发请求重复消费。 */ const take = async (key) => { try { const r = await getRedis(); if (!r) { return null; } if (!(await ensureRedisConnection(r))) { return null; } const value = await withTimeout( r.eval( "local value = redis.call('GET', KEYS[1]); if value then redis.call('DEL', KEYS[1]); end; return value", 1, key ), REDIS_OPERATION_TIMEOUT_MS, null ); return value ? JSON.parse(value) : null; } catch (error) { console.warn('[Redis] TAKE 失败:', error.message); return null; } }; /** * 安全设置缓存,Redis 不可用时静默失败 */ const setex = async (key, ttl, value) => { try { const r = await getRedis(); if (!r) { return false; } if (!(await ensureRedisConnection(r))) { return false; } const stringValue = typeof value === 'string' ? value : JSON.stringify(value); const result = await withTimeout(r.setex(key, ttl, stringValue), REDIS_OPERATION_TIMEOUT_MS, null); return result === 'OK'; } catch (error) { console.warn('[Redis] SETEX 失败:', error.message); return false; } }; /** * 删除缓存 */ const del = async (key) => { try { const r = await getRedis(); if (!r) { return false; } if (!(await ensureRedisConnection(r))) { return false; } await withTimeout(r.del(key), REDIS_OPERATION_TIMEOUT_MS, null); return true; } catch (error) { console.warn('[Redis] DEL 失败:', error.message); return false; } }; /** * 批量删除匹配的 key */ const delPattern = async (pattern) => { try { const r = await getRedis(); if (!r) { return 0; } if (!(await ensureRedisConnection(r))) { return 0; } let cursor = '0'; let deleted = 0; do { const [nextCursor, keys] = await r.scan(cursor, 'MATCH', pattern, 'COUNT', 100); cursor = nextCursor; if (Array.isArray(keys) && keys.length > 0) { deleted += await r.del(...keys); } } while (cursor !== '0'); return deleted; } catch (error) { console.warn('[Redis] DEL PATTERN 失败:', error.message); return 0; } }; /** * 测试连接 */ const test = async () => { try { const r = await getRedis(); if (!r) { return { success: false, message: '未配置 Redis' }; } if (!(await ensureRedisConnection(r))) { return { success: false, message: '连接失败: Redis 当前不可用' }; } const pong = await r.ping(); return { success: pong === 'PONG', message: pong === 'PONG' ? 'Redis 连接成功' : `连接失败: ${pong}` }; } catch (error) { return { success: false, message: `连接失败: ${error.message}` }; } }; /** * 关闭连接 */ const close = async () => { await closeCurrentClient(); console.log('[Redis] 连接已关闭'); }; /** * 应用已保存的运行时配置,无需重启后端进程。 */ const reload = async () => { configCache = { expiresAt: 0, value: null, }; unavailableUntil = 0; await closeCurrentClient(); const client = await getRedis(); if (client) { await ensureRedisConnection(client); } return client; }; module.exports = { getRedis, get, take, setex, del, delPattern, test, close, reload, get connected() { return connected; }, };