338 lines
8.7 KiB
TypeScript
338 lines
8.7 KiB
TypeScript
/**
|
||
* Redis实现 - 生产环境推荐
|
||
*
|
||
* 功能：
|
||
* - 数据持久化（RDB + AOF）
|
||
* - 支持多实例部署
|
||
* - 自动过期（TTL）
|
||
* - 高可用性
|
||
* - 跨服务共享
|
||
*/
|
||
|
||
import Redis from 'ioredis'
|
||
import type { UploadSession } from './upload-sessions'
|
||
|
||
// 初始化Redis连接
|
||
const redisUrl = process.env.REDIS_URL || 'redis://localhost:6379'
|
||
const redis = new Redis(redisUrl, {
|
||
retryStrategy: (times) => {
|
||
// 重试策略：指数退避，最大延迟2秒，最多重试10次
|
||
if (times > 10) {
|
||
console.error('[Redis] 重试次数超过10次，停止重试')
|
||
return null // 停止重试
|
||
}
|
||
const delay = Math.min(times * 50, 2000)
|
||
return delay
|
||
},
|
||
maxRetriesPerRequest: 3,
|
||
enableReadyCheck: true,
|
||
enableOfflineQueue: true, // 启用离线队列，允许在连接恢复前缓存命令
|
||
lazyConnect: false, // 立即连接
|
||
connectTimeout: 5000, // 连接超时5秒
|
||
commandTimeout: 3000, // 命令超时3秒
|
||
// 自动重连配置
|
||
autoResubscribe: true,
|
||
autoResendUnfulfilledCommands: true,
|
||
})
|
||
|
||
// Redis键前缀
|
||
const KEY_PREFIX = 'upload:session:'
|
||
const EXPIRE_TIME = 8 * 60 * 60 // 8小时，覆盖任务-详情跳过 complete 后排队处理的窗口
|
||
|
||
function parsePositiveIntEnv(name: string, fallback: number): number {
|
||
const value = Number.parseInt(process.env[name] || '', 10)
|
||
return Number.isFinite(value) && value > 0 ? value : fallback
|
||
}
|
||
|
||
async function scanSessionKeys(options?: {
|
||
maxKeys?: number
|
||
timeBudgetMs?: number
|
||
}): Promise<string[]> {
|
||
const maxKeys = options?.maxKeys ?? Number.POSITIVE_INFINITY
|
||
const timeBudgetMs = options?.timeBudgetMs ?? Number.POSITIVE_INFINITY
|
||
const scanCount = parsePositiveIntEnv('UPLOAD_SESSION_SCAN_COUNT', 200)
|
||
const keys = new Set<string>()
|
||
const startedAt = Date.now()
|
||
let cursor = '0'
|
||
|
||
do {
|
||
const [nextCursor, batchKeys] = await redis.scan(
|
||
cursor,
|
||
'MATCH',
|
||
`${KEY_PREFIX}*`,
|
||
'COUNT',
|
||
scanCount
|
||
)
|
||
cursor = nextCursor
|
||
for (const key of batchKeys) {
|
||
keys.add(key)
|
||
if (keys.size >= maxKeys) {
|
||
break
|
||
}
|
||
}
|
||
} while (
|
||
cursor !== '0' &&
|
||
keys.size < maxKeys &&
|
||
Date.now() - startedAt <= timeBudgetMs
|
||
)
|
||
|
||
return Array.from(keys)
|
||
}
|
||
|
||
/**
|
||
* 保存上传会话
|
||
*/
|
||
export async function saveUploadSession(
|
||
uploadId: string,
|
||
session: UploadSession
|
||
): Promise<void> {
|
||
try {
|
||
const key = `${KEY_PREFIX}${uploadId}`
|
||
|
||
// 将Map转换为数组以便序列化
|
||
const data = {
|
||
...session,
|
||
chunks: Array.from(session.chunks.entries()),
|
||
}
|
||
|
||
// 使用SETEX设置键值并指定过期时间
|
||
await redis.setex(key, EXPIRE_TIME, JSON.stringify(data))
|
||
|
||
console.log(`[Redis] 保存上传会话: ${uploadId}, 过期时间: ${EXPIRE_TIME}秒`)
|
||
} catch (error) {
|
||
console.error(`[Redis] 保存上传会话失败: ${uploadId}`, error)
|
||
throw error
|
||
}
|
||
}
|
||
|
||
/**
|
||
* 获取上传会话
|
||
*/
|
||
export async function getUploadSession(
|
||
uploadId: string
|
||
): Promise<UploadSession | null> {
|
||
try {
|
||
const key = `${KEY_PREFIX}${uploadId}`
|
||
const data = await redis.get(key)
|
||
|
||
if (!data) {
|
||
return null
|
||
}
|
||
|
||
const session = JSON.parse(data)
|
||
|
||
// 将数组转换回Map
|
||
return {
|
||
...session,
|
||
chunks: new Map(session.chunks || []),
|
||
}
|
||
} catch (error) {
|
||
console.error(`[Redis] 获取上传会话失败: ${uploadId}`, error)
|
||
return null
|
||
}
|
||
}
|
||
|
||
/**
|
||
* 删除上传会话
|
||
*/
|
||
export async function deleteUploadSession(uploadId: string): Promise<void> {
|
||
try {
|
||
const key = `${KEY_PREFIX}${uploadId}`
|
||
await redis.del(key)
|
||
console.log(`[Redis] 删除上传会话: ${uploadId}`)
|
||
} catch (error) {
|
||
console.error(`[Redis] 删除上传会话失败: ${uploadId}`, error)
|
||
throw error
|
||
}
|
||
}
|
||
|
||
/**
|
||
* 更新块信息
|
||
*/
|
||
export async function updateChunk(
|
||
uploadId: string,
|
||
chunkIndex: number,
|
||
chunkPath: string
|
||
): Promise<void> {
|
||
try {
|
||
const session = await getUploadSession(uploadId)
|
||
if (!session) {
|
||
throw new Error(`上传会话不存在: ${uploadId}`)
|
||
}
|
||
|
||
session.chunks.set(chunkIndex, chunkPath)
|
||
await saveUploadSession(uploadId, session)
|
||
|
||
console.log(`[Redis] 更新块信息: ${uploadId}, 块 ${chunkIndex}`)
|
||
} catch (error) {
|
||
console.error(`[Redis] 更新块信息失败: ${uploadId}, 块 ${chunkIndex}`, error)
|
||
throw error
|
||
}
|
||
}
|
||
|
||
/**
|
||
* 检查会话是否存在
|
||
*/
|
||
export async function hasUploadSession(uploadId: string): Promise<boolean> {
|
||
try {
|
||
const key = `${KEY_PREFIX}${uploadId}`
|
||
const exists = await redis.exists(key)
|
||
return exists === 1
|
||
} catch (error) {
|
||
console.error(`[Redis] 检查会话是否存在失败: ${uploadId}`, error)
|
||
return false
|
||
}
|
||
}
|
||
|
||
/**
|
||
* 获取会话的剩余过期时间
|
||
*/
|
||
export async function getSessionTTL(uploadId: string): Promise<number> {
|
||
try {
|
||
const key = `${KEY_PREFIX}${uploadId}`
|
||
const ttl = await redis.ttl(key)
|
||
return ttl
|
||
} catch (error) {
|
||
console.error(`[Redis] 获取会话TTL失败: ${uploadId}`, error)
|
||
return -2 // -2表示键不存在
|
||
}
|
||
}
|
||
|
||
/**
|
||
* 获取所有活跃会话数量
|
||
*/
|
||
export async function getActiveSessionsCount(): Promise<number> {
|
||
try {
|
||
const keys = await scanSessionKeys()
|
||
return keys.length
|
||
} catch (error) {
|
||
console.error('[Redis] 获取活跃会话数量失败', error)
|
||
return 0
|
||
}
|
||
}
|
||
|
||
/**
|
||
* 刷新会话过期时间
|
||
*/
|
||
export async function refreshSessionTTL(uploadId: string): Promise<void> {
|
||
try {
|
||
const key = `${KEY_PREFIX}${uploadId}`
|
||
await redis.expire(key, EXPIRE_TIME)
|
||
console.log(`[Redis] 刷新会话过期时间: ${uploadId}`)
|
||
} catch (error) {
|
||
console.error(`[Redis] 刷新会话过期时间失败: ${uploadId}`, error)
|
||
throw error
|
||
}
|
||
}
|
||
|
||
/**
|
||
* 刷新上传会话（保留已上传进度）
|
||
* 用于会话过期后恢复上传，不清空已上传的分片记录
|
||
*/
|
||
export async function refreshUploadSession(uploadId: string): Promise<boolean> {
|
||
try {
|
||
const key = `${KEY_PREFIX}${uploadId}`
|
||
const exists = await redis.expire(key, EXPIRE_TIME)
|
||
if (exists === 1) {
|
||
console.log(`[Redis] 刷新上传会话成功: ${uploadId}, TTL: ${EXPIRE_TIME}秒`)
|
||
return true
|
||
} else {
|
||
console.warn(`[Redis] 刷新上传会话失败，会话不存在: ${uploadId}`)
|
||
return false
|
||
}
|
||
} catch (error) {
|
||
console.error(`[Redis] 刷新上传会话失败: ${uploadId}`, error)
|
||
return false
|
||
}
|
||
}
|
||
|
||
/**
|
||
* 设置已完成会话的过期时间为 1 分钟（立即清理）
|
||
*/
|
||
export async function setCompletedSessionTTL(uploadId: string): Promise<void> {
|
||
try {
|
||
const key = `${KEY_PREFIX}${uploadId}`
|
||
const COMPLETED_TTL = 60 // 1分钟
|
||
await redis.expire(key, COMPLETED_TTL)
|
||
console.log(`[Redis] 已完成会话，TTL 缩短为 1 分钟: ${uploadId}`)
|
||
} catch (error) {
|
||
console.error(`[Redis] 设置已完成会话 TTL 失败: ${uploadId}`, error)
|
||
throw error
|
||
}
|
||
}
|
||
|
||
/**
|
||
* 获取所有上传会话
|
||
*/
|
||
export async function getAllSessions(): Promise<Map<string, UploadSession>> {
|
||
try {
|
||
const keys = await scanSessionKeys({
|
||
maxKeys: parsePositiveIntEnv('UPLOAD_SESSION_SCAN_MAX_KEYS', 2000),
|
||
timeBudgetMs: parsePositiveIntEnv('UPLOAD_SESSION_SCAN_TIME_BUDGET_MS', 150),
|
||
})
|
||
const sessions = new Map<string, UploadSession>()
|
||
|
||
if (keys.length === 0) {
|
||
return sessions
|
||
}
|
||
|
||
const pipeline = redis.pipeline()
|
||
keys.forEach((key) => pipeline.get(key))
|
||
const results = await pipeline.exec()
|
||
|
||
results?.forEach((result, index) => {
|
||
const key = keys[index]
|
||
const error = result?.[0]
|
||
const data = result?.[1] as string | null
|
||
if (!key || error || !data) {
|
||
if (error && key) {
|
||
console.error(`[Redis] 获取会话失败: ${key}`, error)
|
||
}
|
||
return
|
||
}
|
||
|
||
try {
|
||
const uploadId = key.replace(KEY_PREFIX, '')
|
||
const session = JSON.parse(data)
|
||
sessions.set(uploadId, {
|
||
...session,
|
||
chunks: new Map(session.chunks || []),
|
||
})
|
||
} catch (error) {
|
||
console.error(`[Redis] 解析会话失败: ${key}`, error)
|
||
}
|
||
})
|
||
|
||
return sessions
|
||
} catch (error) {
|
||
console.error('[Redis] 获取所有会话失败', error)
|
||
return new Map()
|
||
}
|
||
}
|
||
|
||
// 错误处理
|
||
redis.on('error', (err) => {
|
||
console.error('[Redis] 连接错误:', err)
|
||
// 不抛出错误，允许应用继续运行（降级到内存Map）
|
||
})
|
||
|
||
redis.on('connect', () => {
|
||
console.log('[Redis] 连接成功')
|
||
})
|
||
|
||
redis.on('ready', () => {
|
||
console.log('[Redis] 准备就绪')
|
||
})
|
||
|
||
redis.on('close', () => {
|
||
console.warn('[Redis] 连接关闭')
|
||
})
|
||
|
||
redis.on('reconnecting', () => {
|
||
console.log('[Redis] 重新连接中...')
|
||
})
|
||
|
||
// 导出Redis实例（用于其他操作）
|
||
export { redis }
|
||
|