Redis最佳实践指南：高性能内存数据库实战技巧

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34

// 原子计数器
const incrementCounter = async (key, increment = 1) => {
    return await redis.incrBy(key, increment);
};

// 设置带TTL的缓存
const setCacheWithTTL = async (key, value, ttl = 3600) => {
    await redis.setex(key, ttl, JSON.stringify(value));
};

// 批量获取
const mGetCached = async (keys) => {
    const values = await redis.mget(keys);
    return values.map(value => value ? JSON.parse(value) : null);
};

// 使用Hash代替多个String（当字段较多时）
const userInfo = {
    name: 'John Doe',
    email: 'john@example.com',
    age: 30,
    city: 'New York'
};

// ❌ 不推荐：多个String键
await redis.mset([
    'user:1:name', userInfo.name,
    'user:1:email', userInfo.email,
    'user:1:age', userInfo.age,
    'user:1:city', userInfo.city
]);

// ✅ 推荐：单个Hash
await redis.hset('user:1', userInfo);

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32

// 用户信息存储
const setUserProfile = async (userId, profile) => {
    const hashKey = `user:${userId}:profile`;

    await redis.hset(hashKey, {
        name: profile.name,
        email: profile.email,
        bio: profile.bio,
        lastUpdate: Date.now()
    });

    // 设置过期时间
    await redis.expire(hashKey, 86400); // 24小时
};

// 获取部分字段
const getUserFields = async (userId, fields) => {
    const hashKey = `user:${userId}:profile`;
    return await redis.hmget(hashKey, fields);
};

// 更新单个字段
const updateUserField = async (userId, field, value) => {
    const hashKey = `user:${userId}:profile`;
    await redis.hset(hashKey, field, value);
    await redis.hset(hashKey, 'lastUpdate', Date.now());
};

// Hash计数器（适用于小规模计数）
const hashIncrement = async (key, field, increment = 1) => {
    return await redis.hincrby(key, field, increment);
};

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60

// 消息队列（简单实现）
class MessageQueue {
    constructor(redis, queueName) {
        this.redis = redis;
        this.queueName = queueName;
        this.processingName = `${queueName}:processing`;
    }

    async enqueue(message) {
        return await this.redis.lpush(this.queueName, JSON.stringify(message));
    }

    async dequeue() {
        return await this.redis.brpoplpush(
            this.queueName,
            this.processingName,
            0
        );
    }

    async complete(messageId) {
        // 从处理队列中移除
        return await this.redis.lrem(this.processingName, 1, messageId);
    }

    async getQueueLength() {
        return await this.redis.llen(this.queueName);
    }

    async getProcessingLength() {
        return await this.redis.llen(this.processingName);
    }
}

// 时间线（最新消息）
const addTimelineEvent = async (userId, event) => {
    const key = `timeline:${userId}`;
    const eventData = {
        id: generateId(),
        type: event.type,
        data: event.data,
        timestamp: Date.now()
    };

    // 添加到时间线头部
    await redis.lpush(key, JSON.stringify(eventData));

    // 保持最近100条消息
    await redis.ltrim(key, 0, 99);

    // 设置过期时间
    await redis.expire(key, 86400 * 7); // 7天
};

const getTimeline = async (userId, limit = 20) => {
    const key = `timeline:${userId}`;
    const events = await redis.lrange(key, 0, limit - 1);

    return events.map(event => JSON.parse(event));
};

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67

// 用户标签系统
class UserTags {
    constructor(redis) {
        this.redis = redis;
    }

    async addTag(userId, tag) {
        const key = `user:${userId}:tags`;
        await redis.sadd(key, tag);
        return await redis.expire(key, 86400 * 30); // 30天
    }

    async removeTag(userId, tag) {
        const key = `user:${userId}:tags`;
        return await redis.srem(key, tag);
    }

    async getUserTags(userId) {
        const key = `user:${userId}:tags`;
        return await redis.smembers(key);
    }

    async getTagUsers(tag) {
        const key = `tag:${tag}:users`;
        return await redis.smembers(key);
    }

    // 双向索引：同时维护用户标签和标签用户
    async addTagToUser(userId, tag) {
        const userTagsKey = `user:${userId}:tags`;
        const tagUsersKey = `tag:${tag}:users`;

        await this.redis.multi()
            .sadd(userTagsKey, tag)
            .sadd(tagUsersKey, userId)
            .expire(userTagsKey, 86400 * 30)
            .expire(tagUsersKey, 86400 * 30)
            .exec();
    }
}

// 集合操作：找出共同标签
const findCommonTags = async (userIds) => {
    const keys = userIds.map(id => `user:${id}:tags`);

    // 如果只有一个用户，返回其所有标签
    if (keys.length === 1) {
        return await this.redis.smembers(keys[0]);
    }

    // 使用SINTER找出共同标签
    return await this.redis.sinter(keys);
};

// 推荐相似用户
const findSimilarUsers = async (userId, threshold = 0.3) => {
    const userTagsKey = `user:${userId}:tags`;
    const userTags = await this.redis.smembers(userTagsKey);

    if (userTags.length === 0) return [];

    const similarUsers = [];

    // 这里需要获取所有用户标签进行比较
    // 在实际应用中，可能需要使用其他数据结构来优化
    return similarUsers;
};

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133

// 排行榜系统
class Leaderboard {
    constructor(redis, leaderboardName) {
        this.redis = redis;
        this.leaderboardName = leaderboardName;
        this.ttl = 86400 * 7; // 7天
    }

    async addScore(member, score, memberData = {}) {
        // 添加分数
        await this.redis.zadd(this.leaderboardName, score, member);

        // 存储成员详细信息
        if (Object.keys(memberData).length > 0) {
            const dataKey = `${this.leaderboardName}:data:${member}`;
            await this.redis.hset(dataKey, memberData);
            await this.redis.expire(dataKey, this.ttl);
        }

        // 设置过期时间
        await this.redis.expire(this.leaderboardName, this.ttl);
    }

    async getTopRank(limit = 10) {
        // 获取排行榜前N名
        const topUsers = await this.redis.zrevrange(
            this.leaderboardName,
            0,
            limit - 1,
            'WITHSCORES'
        );

        const result = [];

        for (let i = 0; i < topUsers.length; i += 2) {
            const member = topUsers[i];
            const score = topUsers[i + 1];
            const rank = i / 2 + 1;

            // 获取用户详细信息
            const dataKey = `${this.leaderboardName}:data:${member}`;
            const userData = await this.redis.hgetall(dataKey);

            result.push({
                rank,
                member,
                score: parseFloat(score),
                ...userData
            });
        }

        return result;
    }

    async getUserRank(member) {
        const rank = await this.redis.zrevrank(this.leaderboardName, member);
        return rank !== null ? rank + 1 : null;
    }

    async getScore(member) {
        return await this.redis.zscore(this.leaderboardName, member);
    }

    async updateScore(member, newScore) {
        return await this.redis.zadd(this.leaderboardName, newScore, member);
    }
}

// 延迟队列（Sorted Set实现）
class DelayedQueue {
    constructor(redis, queueName) {
        this.redis = redis;
        this.queueName = queueName;
        this.processing = false;
    }

    async addTask(taskId, taskData, delaySeconds) {
        const executeAt = Date.now() + delaySeconds * 1000;
        const task = {
            id: taskId,
            data: taskData,
            executeAt,
            addedAt: Date.now()
        };

        return await this.redis.zadd(
            this.queueName,
            executeAt,
            JSON.stringify(task)
        );
    }

    async processNext() {
        if (this.processing) return;

        this.processing = true;

        try {
            const now = Date.now();

            // 获取到期的任务
            const tasks = await this.redis.zrangebyscore(
                this.queueName,
                0,
                now,
                'LIMIT',
                1
            );

            if (tasks.length === 0) {
                this.processing = false;
                return null;
            }

            const task = JSON.parse(tasks[0]);

            // 从队列中移除
            await this.redis.zrem(this.queueName, tasks[0]);

            // 处理任务
            await this.handleTask(task);

            return task;
        } finally {
            this.processing = false;
        }
    }

    async handleTask(task) {
        console.log(`Processing task ${task.id}:`, task.data);
        // 实现具体的任务处理逻辑
    }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37

// 批量操作优化
class RedisBatch {
    constructor(redis) {
        this.redis = redis;
    }

    async batchSet(keyValues) {
        const pipeline = this.redis.pipeline();

        for (const [key, value] of Object.entries(keyValues)) {
            pipeline.set(key, JSON.stringify(value));
        }

        return await pipeline.exec();
    }

    async batchUpdateScores(leaderboard, updates) {
        const pipeline = this.redis.pipeline();

        for (const [member, score] of updates) {
            pipeline.zadd(leaderboard, score, member);
        }

        return await pipeline.exec();
    }

    // 原子性批量操作
    async atomicUpdate(userData) {
        const key = `user:${userData.id}`;

        return await this.redis.multi()
            .hset(key, userData)
            .expire(key, 3600)
            .sadd('users', userData.id)
            .exec();
    }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61

// Key设计最佳实践
class KeyNaming {
    // 使用冒号分隔的层级结构
    static user(userId) {
        return `user:${userId}`;
    }

    static userProfile(userId) {
        return `user:${userId}:profile`;
    }

    static userSessions(userId) {
        return `user:${userId}:sessions`;
    }

    static cache(namespace, key) {
        return `cache:${namespace}:${key}`;
    }

    static lock(resource) {
        return `lock:${resource}`;
    }

    static counter(name) {
        return `counter:${name}`;
    }
}

// 压缩存储
const compressData = (data) => {
    if (typeof data === 'string') {
        return data; // 已经是字符串
    }
    return JSON.stringify(data);
};

// 选择合适的数据结构
const chooseDataStructure = (useCase) => {
    switch (useCase) {
        case 'simpleCache':
            return 'String'; // 简单键值缓存

        case 'userProfile':
            return 'Hash'; // 用户档案包含多个字段

        case 'messageQueue':
            return 'List'; // 消息队列

        case 'userTags':
            return 'Set'; // 用户标签集合

        case 'leaderboard':
            return 'Sorted Set'; // 排行榜

        case 'rateLimit':
            return 'String + Expire'; // 限流器

        default:
            return 'String';
    }
};

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100

class MultiLevelCache {
    constructor(redis, localCache) {
        this.redis = redis;
        this.localCache = localCache; // 例如：Map或LRU Cache
    }

    async get(key) {
        // L1：本地缓存
        let value = this.localCache.get(key);
        if (value !== undefined) {
            return value;
        }

        // L2：Redis缓存
        value = await this.redis.get(key);
        if (value !== null) {
            const parsedValue = JSON.parse(value);
            // 回填本地缓存
            this.localCache.set(key, parsedValue, 300); // 5分钟
            return parsedValue;
        }

        // L3：数据库或其他存储
        value = await this.fetchFromDatabase(key);
        if (value !== null) {
            // 写入所有缓存层
            await this.set(key, value);
        }

        return value;
    }

    async set(key, value, ttl = 3600) {
        // 同时写入所有缓存层
        this.localCache.set(key, value, Math.min(ttl, 300)); // 本地缓存最多5分钟
        await this.redis.setex(key, ttl, JSON.stringify(value));
    }

    async invalidate(key) {
        this.localCache.delete(key);
        await this.redis.del(key);
    }

    async fetchFromDatabase(key) {
        // 实际的数据库查询逻辑
        console.log(`Fetching ${key} from database`);
        return null; // 示例
    }
}

// 缓存穿透保护
const cacheWithProtection = async (key, fetcher) => {
    // 检查是否存在"空"缓存
    const nullKey = `${key}:null`;
    const isNull = await redis.exists(nullKey);

    if (isNull) {
        return null; // 已知为空
    }

    let value = await redis.get(key);

    if (value !== null) {
        return JSON.parse(value);
    }

    // 并发控制
    const lockKey = `lock:${key}`;
    const lockValue = generateUUID();
    const lockAcquired = await redis.setnx(lockKey, lockValue, 10); // 10秒锁

    if (lockAcquired) {
        try {
            value = await fetcher();

            if (value === null) {
                // 设置"空"缓存，防止穿透
                await redis.setex(nullKey, 60, 'null'); // 1分钟
            } else {
                await redis.setex(key, 3600, JSON.stringify(value));
            }

            return value;
        } finally {
            // 释放锁
            const script = `
                if redis.call("get", KEYS[1]) == ARGV[1] then
                    return redis.call("del", KEYS[1])
                else
                    return 0
                end
            `;
            await redis.eval(script, 1, lockKey, lockValue);
        }
    } else {
        // 等待锁释放后重试
        await new Promise(resolve => setTimeout(resolve, 100));
        return cacheWithProtection(key, fetcher);
    }
};

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103

// Cache-Aside模式
class CacheAside {
    constructor(redis) {
        this.redis = redis;
    }

    async get(key, fetcher) {
        let data = await this.redis.get(key);

        if (data !== null) {
            return JSON.parse(data);
        }

        data = await fetcher();

        if (data !== null) {
            await this.redis.setex(key, 3600, JSON.stringify(data));
        }

        return data;
    }

    async update(key, data, ttl = 3600) {
        // 先更新数据库
        await this.updateDatabase(key, data);

        // 然后更新缓存
        await this.redis.setex(key, ttl, JSON.stringify(data));
    }

    async invalidate(key) {
        await this.redis.del(key);
    }
}

// Write-Through模式
class WriteThrough {
    constructor(redis) {
        this.redis = redis;
    }

    async set(key, data, ttl = 3600) {
        // 同时写入数据库和缓存
        await Promise.all([
            this.updateDatabase(key, data),
            this.redis.setex(key, ttl, JSON.stringify(data))
        ]);
    }
}

// Write-Behind模式
class WriteBehind {
    constructor(redis) {
        this.redis = redis;
        this.writeQueue = [];
        this.batchSize = 100;
        this.flushInterval = 5000; // 5秒
        this.startFlushTimer();
    }

    async set(key, data, ttl = 3600) {
        // 立即写入缓存
        await this.redis.setex(key, ttl, JSON.stringify(data));

        // 加入写队列
        this.writeQueue.push({
            key,
            data,
            timestamp: Date.now()
        });

        // 如果队列满了，立即刷新
        if (this.writeQueue.length >= this.batchSize) {
            await this.flushQueue();
        }
    }

    async flushQueue() {
        if (this.writeQueue.length === 0) return;

        const batch = this.writeQueue.splice(0);

        try {
            // 批量写入数据库
            await this.batchWriteToDatabase(batch);
        } catch (error) {
            console.error('Write-behind batch failed:', error);
            // 重新加入队列
            this.writeQueue.unshift(...batch);
        }
    }

    startFlushTimer() {
        setInterval(() => {
            this.flushQueue();
        }, this.flushInterval);
    }

    async batchWriteToDatabase(batch) {
        // 实现批量数据库写入逻辑
        console.log(`Writing ${batch.length} items to database`);
    }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92

// 简单分布式锁
class DistributedLock {
    constructor(redis) {
        this.redis = redis;
    }

    async acquire(key, ttl = 30) {
        const lockValue = generateUUID();
        const result = await this.redis.setnx(key, lockValue, ttl);

        if (result === 1) {
            return {
                acquired: true,
                value: lockValue
            };
        }

        return {
            acquired: false,
            value: null
        };
    }

    async release(key, lockValue) {
        const script = `
            if redis.call("get", KEYS[1]) == ARGV[1] then
                return redis.call("del", KEYS[1])
            else
                return 0
            end
        `;

        const result = await this.redis.eval(script, 1, key, lockValue);
        return result === 1;
    }

    async extend(key, lockValue, ttl) {
        const script = `
            if redis.call("get", KEYS[1]) == ARGV[1] then
                return redis.call("expire", KEYS[1], ARGV[2])
            else
                return 0
            end
        `;

        return await this.redis.eval(script, 1, key, lockValue, ttl);
    }
}

// 可重入锁
class ReentrantLock {
    constructor(redis) {
        this.redis = redis;
    }

    async acquire(key, lockId, ttl = 30) {
        const lockKey = `reentrant:${key}:${lockId}`;
        const counterKey = `reentrant:${key}:counter`;

        const script = `
            local counter = redis.call("incr", KEYS[2])
            redis.call("expire", KEYS[2], ARGV[1])
            redis.call("set", KEYS[1], "1", "EX", ARGV[1])
            return counter
        `;

        return await this.redis.eval(script, 2, lockKey, counterKey, ttl);
    }

    async release(key, lockId) {
        const lockKey = `reentrant:${key}:${lockId}`;
        const counterKey = `reentrant:${key}:counter`;

        const script = `
            local count = redis.call("get", KEYS[1])
            if count == "0" then
                return 0
            end

            redis.call("decr", KEYS[2])

            if tonumber(redis.call("get", KEYS[2])) == 0 then
                redis.call("del", KEYS[1])
                redis.call("del", KEYS[2])
            end

            return 1
        `;

        return await this.redis.eval(script, 2, lockKey, counterKey);
    }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91

class RedisMonitor {
    constructor(redis) {
        this.redis = redis;
    }

    async getMetrics() {
        const info = await this.redis.info();
        const lines = info.split('\r\n');

        const metrics = {};

        for (const line of lines) {
            if (line.includes(':')) {
                const [key, value] = line.split(':');
                metrics[key] = value;
            }
        }

        return {
            connected_clients: parseInt(metrics.connected_clients) || 0,
            used_memory: metrics.used_memory_human || '0B',
            total_commands_processed: parseInt(metrics.total_commands_processed) || 0,
            keyspace_hits: parseInt(metrics.keyspace_hits) || 0,
            keyspace_misses: parseInt(metrics.keyspace_misses) || 0,
            instantaneous_ops_per_sec: parseFloat(metrics.instantaneous_ops_per_sec) || 0,
            memory_usage_ratio: parseFloat(metrics.mem_fragmentation_ratio) || 0
        };
    }

    async getSlowLog(count = 10) {
        return await this.redis.slowlog('GET', count);
    }

    async getMemoryUsage() {
        const memoryInfo = await this.redis.memory('usage');

        return {
            peak: memoryInfo.peak,
            used: memoryInfo.used,
            rss: memoryInfo.rss,
            overhead: memoryInfo.overhead,
            startup: memoryInfo.startup,
            dataset: memoryInfo.dataset
        };
    }

    async checkHealth() {
        try {
            await this.redis.ping();

            const metrics = await this.getMetrics();

            return {
                status: 'healthy',
                timestamp: new Date().toISOString(),
                metrics
            };
        } catch (error) {
            return {
                status: 'unhealthy',
                timestamp: new Date().toISOString(),
                error: error.message
            };
        }
    }

    // 定期监控
    startMonitoring(interval = 60000) { // 1分钟
        setInterval(async () => {
            const health = await this.checkHealth();

            if (health.status === 'unhealthy') {
                console.error('Redis health check failed:', health.error);
                // 发送告警
                this.sendAlert(health);
            }

            // 记录指标
            console.log('Redis metrics:', health.metrics);
        }, interval);
    }

    sendAlert(health) {
        // 实现告警逻辑
        console.error('Redis Alert:', health);
    }
}

// 使用示例
const monitor = new RedisMonitor(redis);
monitor.startMonitoring();

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70

// 慢查询检测
class SlowQueryDetector {
    constructor(redis, threshold = 100) {
        this.redis = redis;
        this.threshold = threshold;
        this.slowQueries = [];
    }

    async detectSlowQuery(command, startTime, endTime) {
        const duration = endTime - startTime;

        if (duration > this.threshold) {
            const slowQuery = {
                command,
                duration,
                timestamp: new Date().toISOString()
            };

            this.slowQueries.push(slowQuery);

            console.warn(`Slow query detected: ${command} took ${duration}ms`);

            // 发送到监控系统
            this.reportSlowQuery(slowQuery);

            // 保持最近100个慢查询
            if (this.slowQueries.length > 100) {
                this.slowQueries.shift();
            }
        }
    }

    async reportSlowQuery(query) {
        // 实现慢查询报告逻辑
    }

    getSlowQueries() {
        return this.slowQueries;
    }

    // 装饰器函数
    wrapRedisFunction(originalFunction, functionName) {
        return async (...args) => {
            const startTime = Date.now();

            try {
                const result = await originalFunction.apply(this.redis, args);
                const endTime = Date.now();

                await this.detectSlowQuery(
                    `${functionName}(${args.join(', ')})`,
                    startTime,
                    endTime
                );

                return result;
            } catch (error) {
                const endTime = Date.now();

                await this.detectSlowQuery(
                    `${functionName}(${args.join(', ')}) - ERROR`,
                    startTime,
                    endTime
                );

                throw error;
            }
        };
    }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68

// Redis集群客户端
class RedisClusterClient {
    constructor(nodes) {
        this.nodes = nodes;
        this.connections = new Map();
        this.keySlotCount = 16384;
        this.initializeConnections();
    }

    initializeConnections() {
        for (const node of this.nodes) {
            const redis = new Redis({
                host: node.host,
                port: node.port,
                retryDelayOnFailover: 100,
                enableReadyCheck: true,
                maxRetriesPerRequest: 3
            });

            this.connections.set(`${node.host}:${node.port}`, redis);
        }
    }

    getSlot(key) {
        return this.crc16(key) % this.keySlotCount;
    }

    getNodeForSlot(slot) {
        // 简化版本，实际应该根据slot范围选择节点
        const nodeIndex = Math.floor(slot / (this.keySlotCount / this.nodes.length));
        return this.nodes[nodeIndex];
    }

    getConnectionForKey(key) {
        const slot = this.getSlot(key);
        const node = this.getNodeForSlot(slot);
        return this.connections.get(`${node.host}:${node.port}`);
    }

    async get(key) {
        const connection = this.getConnectionForKey(key);
        return await connection.get(key);
    }

    async set(key, value, ttl) {
        const connection = this.getConnectionForKey(key);

        if (ttl) {
            return await connection.setex(key, ttl, value);
        } else {
            return await connection.set(key, value);
        }
    }

    crc16(str) {
        let crc = 0xFFFF;

        for (let i = 0; i < str.length; i++) {
            crc = ((crc << 8) & 0xFF00) | (crc >> 8);
            crc ^= str.charCodeAt(i) & 0xFF;
            crc ^= ((crc & 0xFF) << 4) << 8;
            crc = ((crc << 8) & 0xFF00) | (crc >> 8);
            crc ^= ((crc & 0xFF) >> 4) << 8;
        }

        return crc ^ 0xFFFF;
    }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91

class RedisFailover {
    constructor(primaryConfig, replicaConfigs) {
        this.primaryConfig = primaryConfig;
        this.replicaConfigs = replicaConfigs;
        this.currentPrimary = null;
        this.replicas = [];
        this.initializeConnections();
    }

    async initializeConnections() {
        try {
            // 尝试连接主服务器
            this.currentPrimary = new Redis(this.primaryConfig);
            await this.currentPrimary.ping();
            console.log('Connected to primary server');
        } catch (error) {
            console.error('Primary server unavailable:', error);
            await this.promoteReplica();
        }

        // 连接从服务器
        for (const config of this.replicaConfigs) {
            try {
                const replica = new Redis(config);
                await replica.ping();
                this.replicas.push(replica);
                console.log(`Connected to replica server: ${config.host}:${config.port}`);
            } catch (error) {
                console.warn(`Replica server unavailable: ${config.host}:${config.port}`);
            }
        }
    }

    async promoteReplica() {
        for (let i = 0; i < this.replicas.length; i++) {
            try {
                // 尝试提升从服务器为主服务器
                const replica = this.replicas[i];
                await replica.replica('NO ONE');

                // 等待从服务器完成角色切换
                await new Promise(resolve => setTimeout(resolve, 1000));

                this.currentPrimary = replica;
                console.log(`Promoted replica ${i} to primary`);

                // 重新配置其他从服务器
                await this.reconfigureReplicas();

                break;
            } catch (error) {
                console.error(`Failed to promote replica ${i}:`, error);
            }
        }

        if (!this.currentPrimary) {
            throw new Error('No available replicas to promote');
        }
    }

    async reconfigureReplicas() {
        for (const replica of this.replicas) {
            if (replica !== this.currentPrimary) {
                try {
                    await replica.replica(this.primaryConfig.host, this.primaryConfig.port);
                } catch (error) {
                    console.error('Failed to reconfigure replica:', error);
                }
            }
        }
    }

    async handleFailover() {
        if (this.currentPrimary) {
            try {
                await this.currentPrimary.ping();
                return; // 主服务器正常
            } catch (error) {
                console.error('Primary server failed:', error);
            }
        }

        await this.promoteReplica();
    }

    startHealthCheck() {
        setInterval(async () => {
            await this.handleFailover();
        }, 5000); // 每5秒检查一次
    }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85

class RedisBackup {
    constructor(redis) {
        this.redis = redis;
    }

    async createSnapshot(backupDir) {
        const timestamp = new Date().toISOString().replace(/[:.]/g, '-');
        const filename = `redis-backup-${timestamp}.rdb`;
        const filePath = path.join(backupDir, filename);

        try {
            // 触发Redis RDB快照
            await this.redis.save();

            // 等待快照完成
            await this.waitForSaveCompletion();

            // 复制快照文件到备份目录
            await this.copyRdbFile(filePath);

            console.log(`Backup created: ${filePath}`);

            return filename;
        } catch (error) {
            console.error('Backup failed:', error);
            throw error;
        }
    }

    async waitForSaveCompletion() {
        let saveInProgress = true;

        while (saveInProgress) {
            const info = await this.redis.info('persistence');

            if (info.includes('saving: 0')) {
                saveInProgress = false;
            } else {
                await new Promise(resolve => setTimeout(resolve, 1000));
            }
        }
    }

    async copyRdbFile(destination) {
        const rdbPath = '/var/lib/redis/dump.rdb'; // 默认RDB路径

        try {
            await fs.copyFile(rdbPath, destination);
        } catch (error) {
            throw new Error(`Failed to copy RDB file: ${error.message}`);
        }
    }

    async scheduleBackups(backupDir, interval = 24 * 60 * 60 * 1000) { // 24小时
        setInterval(async () => {
            try {
                await this.createSnapshot(backupDir);

                // 清理旧备份（保留最近7天）
                await this.cleanupOldBackups(backupDir, 7);
            } catch (error) {
                console.error('Scheduled backup failed:', error);
            }
        }, interval);
    }

    async cleanupOldBackups(backupDir, daysToKeep) {
        const files = await fs.readdir(backupDir);
        const now = Date.now();

        for (const file of files) {
            if (file.startsWith('redis-backup-') && file.endsWith('.rdb')) {
                const filePath = path.join(backupDir, file);
                const stats = await fs.stat(filePath);

                const ageInDays = (now - stats.mtime.getTime()) / (24 * 60 * 60 * 1000);

                if (ageInDays > daysToKeep) {
                    await fs.unlink(filePath);
                    console.log(`Removed old backup: ${file}`);
                }
            }
        }
    }
}

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101

// TLS连接配置
class SecureRedisClient {
    constructor(config) {
        this.config = {
            host: config.host || 'localhost',
            port: config.port || 6379,
            tls: config.tls || {},
            password: config.password,
            database: config.database || 0,
            retryDelayOnFailover: 100,
            maxRetriesPerRequest: 3,
            enableReadyCheck: true,
            lazyConnect: true
        };
    }

    createClient() {
        const client = new Redis(this.config);

        // 连接错误处理
        client.on('error', (err) => {
            console.error('Redis Client Error:', err);
        });

        client.on('connect', () => {
            console.log('Redis Client Connected');
        });

        client.on('reconnecting', () => {
            console.log('Redis Client Reconnecting...');
        });

        return client;
    }
}

// 连接池管理
class RedisConnectionPool {
    constructor(config, poolSize = 10) {
        this.config = config;
        this.poolSize = poolSize;
        this.connections = [];
        this.waitingQueue = [];
        this.initializePool();
    }

    async initializePool() {
        for (let i = 0; i < this.poolSize; i++) {
            const client = new Redis(this.config);
            this.connections.push({
                client,
                inUse: false,
                id: i
            });
        }
    }

    async getConnection() {
        return new Promise((resolve, reject) => {
            // 查找可用连接
            const available = this.connections.find(conn => !conn.inUse);

            if (available) {
                available.inUse = true;
                resolve(available.client);
            } else {
                // 加入等待队列
                this.waitingQueue.push({ resolve, reject });
            }
        });
    }

    releaseConnection(client) {
        const connection = this.connections.find(conn => conn.client === client);

        if (connection) {
            connection.inUse = false;

            // 处理等待队列
            if (this.waitingQueue.length > 0) {
                const waiter = this.waitingQueue.shift();
                connection.inUse = true;
                waiter.resolve(connection.client);
            }
        }
    }

    // 连接健康检查
    async healthCheck() {
        const client = await this.getConnection();

        try {
            await client.ping();
            this.releaseConnection(client);
            return { status: 'healthy' };
        } catch (error) {
            this.releaseConnection(client);
            return { status: 'unhealthy', error: error.message };
        }
    }
}