redis：Redis Lua脚本深度解析

WeiLai1112

993人浏览 · 2025-07-03 12:50:51

WeiLai1112 · 2025-07-03 12:50:51 发布

Redis Lua脚本深度解析：原理、实践与大厂应用

一、Redis Lua脚本核心原理

Redis Lua脚本功能允许开发者将多个Redis命令组合成一个原子操作在服务端执行。其核心特点包括：

原子性：整个脚本作为一个命令执行，执行期间不会插入其他命令
高性能：减少网络往返时延，特别适合复杂多命令操作
灵活性：可使用Lua语言的控制结构和Redis命令组合

二、Lua脚本在分布式锁中的应用

1. 分布式锁实现时序

2. 电商秒杀项目实战

在某电商平台秒杀系统中，我们使用Lua脚本实现库存扣减的原子操作：

// 库存扣减Lua脚本
String STOCK_DEDUCTION_SCRIPT =
    "local stock = tonumber(redis.call('get', KEYS[1])) " +
    "if stock >= tonumber(ARGV[1]) then " +
    "    redis.call('decrby', KEYS[1], ARGV[1]) " +
    "    redis.call('sadd', KEYS[2], ARGV[2]) " +
    "    return 1 " +
    "else " +
    "    return 0 " +
    "end";

// 预加载脚本
String scriptSha = redisTemplate.scriptLoad(STOCK_DEDUCTION_SCRIPT);

// 执行脚本
public boolean deductStock(String productId, int quantity, String userId) {
    List<String> keys = Arrays.asList("stock:" + productId, "order:" + productId);
    Object result = redisTemplate.execute(
        new DefaultRedisScript<>(STOCK_DEDUCTION_SCRIPT, Long.class),
        keys, String.valueOf(quantity), userId);
    return ((Long)result) == 1L;
}

性能对比：

普通多命令操作：平均耗时15ms（网络往返2次）
Lua脚本操作：平均耗时3ms（网络往返1次）
QPS提升：从2000提升到8000+

三、大厂面试深度追问与解决方案

追问1：如何保证Lua脚本的原子性？Redis崩溃时脚本会怎样？

问题分析：
Redis单线程模型保证了脚本执行的原子性，但需要考虑崩溃恢复场景。

解决方案：

原子性保障机制：

Redis使用单线程处理命令，脚本执行期间不会处理其他请求
脚本中的所有命令要么全部执行，要么全部不执行

崩溃恢复方案：

生产环境最佳实践：

// 带幂等控制的脚本执行
public Object executeScriptSafely(RedisScript<?> script, List<String> keys, Object... args) {
    int maxRetries = 3;
    for (int i = 0; i < maxRetries; i++) {
        try {
            return redisTemplate.execute(script, keys, args);
        } catch (RedisConnectionFailureException e) {
            // 检查数据库确认状态
            boolean completed = transactionDao.checkStatus(keys.get(0));
            if (completed) {
                return null; // 已成功无需重试
            }
            if (i == maxRetries - 1) {
                throw e;
            }
            Thread.sleep(100 * (i + 1));
        }
    }
    return null;
}

监控与告警：

# 监控脚本执行情况
redis-cli info commandstats | grep eval
redis-cli info persistence | grep rdb_last_bgsave_status

追问2：如何处理Lua脚本中的大Key问题？如何优化执行性能？

问题场景：
当脚本操作大Value（如10MB的Hash）时，会导致执行阻塞，影响Redis整体性能。

解决方案：

大Key拆分方案：

-- 将大Hash拆分为多个子Hash
local function processBigHash(keyPrefix, fields)
    local results = {}
    for i, field in ipairs(fields) do
        local subKey = keyPrefix .. (i % 10) -- 拆分为10个子Key
        table.insert(results, redis.call('HGET', subKey, field))
    end
    return results
end

性能优化技巧：

// 使用SCRIPT KILL终止长时间运行的脚本
public void monitorScriptTimeout(Duration timeout) {
    ScheduledExecutorService scheduler = Executors.newSingleThreadScheduledExecutor();
    scheduler.schedule(() -> {
        Long currentScript = redisTemplate.execute(
            "return redis.call('SCRIPT', 'KILL')", Collections.emptyList());
        if (currentScript != null) {
            log.warn("Killed long running script");
        }
    }, timeout.toMillis(), TimeUnit.MILLISECONDS);
}

// 分批次处理
String BATCH_PROCESS_SCRIPT =
    "local cursor = tonumber(ARGV[1]) " +
    "local limit = tonumber(ARGV[2]) " +
    "local result = {} " +
    "local nextCursor = redis.call('SCAN', cursor, 'COUNT', limit) " +
    "for i, key in ipairs(nextCursor[2]) do " +
    "    table.insert(result, redis.call('GET', key)) " +
    "end " +
    "return {nextCursor[1], result}";

内存优化方案：

-- 使用游标增量处理
local cursor = 0
local results = {}
repeat
    local reply = redis.call('SCAN', cursor, 'MATCH', ARGV[1], 'COUNT', 1000)
    cursor = tonumber(reply[1])
    for _, key in ipairs(reply[2]) do
        -- 处理每个key
        table.insert(results, redis.call('GET', key))
    end
until cursor == 0
return results

生产环境配置：

# 限制脚本执行时间(单位：毫秒)
config set lua-time-limit 5000

# 监控脚本执行
redis-cli info commandstats | grep -E "eval|script"

四、高级应用场景

1. 分布式限流器实现

-- 令牌桶限流算法
local tokens_key = KEYS[1]
local timestamp_key = KEYS[2]
local rate = tonumber(ARGV[1])
local capacity = tonumber(ARGV[2])
local now = tonumber(ARGV[3])
local requested = tonumber(ARGV[4])

local fill_time = capacity/rate
local ttl = math.floor(fill_time*2)

local last_tokens = tonumber(redis.call("get", tokens_key))
if last_tokens == nil then
    last_tokens = capacity
end

local last_refreshed = tonumber(redis.call("get", timestamp_key))
if last_refreshed == nil then
    last_refreshed = 0
end

local delta = math.max(0, now-last_refreshed)
local filled_tokens = math.min(capacity, last_tokens+(delta*rate))
local allowed = filled_tokens >= requested
local new_tokens = filled_tokens
if allowed then
    new_tokens = filled_tokens - requested
end

redis.call("setex", tokens_key, ttl, new_tokens)
redis.call("setex", timestamp_key, ttl, now)

return { allowed, new_tokens }

2. 热点数据发现与处理

// 热点Key探测脚本
String HOT_KEY_DETECT_SCRIPT =
    "local counts = {} " +
    "local keys = redis.call('SCAN', 0, 'MATCH', ARGV[1], 'COUNT', 1000)[2] " +
    "for _, key in ipairs(keys) do " +
    "    local count = redis.call('PFCOUNT', 'access:'..key) " +
    "    if count > tonumber(ARGV[2]) then " +
    "        table.insert(counts, {key, count}) " +
    "    end " +
    "end " +
    "return counts";

public List<Pair<String, Long>> detectHotKeys(String pattern, long threshold) {
    return (List<Pair<String, Long>>) redisTemplate.execute(
        new DefaultRedisScript<>(HOT_KEY_DETECT_SCRIPT, List.class),
        Collections.emptyList(), pattern, String.valueOf(threshold));
}