Lua 脚本原子执行

# Lua 脚本原子执行

1. Lua 脚本基础#

1.1 为什么需要 Lua 脚本#

Redis 执行 Lua 脚本时，整个脚本是原子性的——脚本执行期间不会插入其他客户端命令：

普通方式:
  GET key → 应用层计算 → SET key  ← 中间可能被其他客户端修改

Lua 脚本:
  EVAL "local v = redis.call('GET', KEYS[1]); ..." 1 key
  ← 整个过程原子执行，不会被中断

1.2 EVAL 命令#

EVAL script numkeys key [key ...] arg [arg ...]

-- script: Lua 脚本
-- numkeys: 键的数量
-- key: 键名列表（通过 KEYS[1], KEYS[2]... 访问）
-- arg: 参数列表（通过 ARGV[1], ARGV[2]... 访问）

1.3 基本示例#

-- 简单的 GET + SET
EVAL "redis.call('SET', KEYS[1], ARGV[1]); return 'OK'" 1 mykey myvalue

-- 限流器
EVAL "
  local count = redis.call('INCR', KEYS[1])
  if count == 1 then
    redis.call('EXPIRE', KEYS[1], ARGV[1])
  end
  if count > tonumber(ARGV[2]) then
    return 0
  end
  return 1
" 1 rate_limit:user1 60 100

2. redis.call 与 redis.pcall#

2.1 区别#

-- redis.call: 错误时脚本终止
local val = redis.call('INCR', 'non_numeric_key')  -- 如果key不是整数，报错终止

-- redis.pcall: 错误时返回错误表
local result = redis.pcall('INCR', 'non_numeric_key')
if type(result) == 'table' and result.err then
    -- 处理错误
    redis.call('SET', 'error_log', result.err)
end

2.2 返回值类型映射#

3. EVALSHA 与脚本缓存#

3.1 脚本缓存机制#

1. 首次 EVAL: 脚本被计算 SHA1 并缓存
2. 后续 EVALSHA: 只发送 SHA1，减少网络传输

SHA1 = SHA1(script)

3.2 EVALSHA 使用#

-- 加载脚本到缓存
SCRIPT LOAD "return redis.call('GET', KEYS[1])"
-- 返回: "a1b2c3d4..." (SHA1)

-- 使用 SHA1 执行
EVALSHA a1b2c3d4... 1 mykey

3.3 脚本缓存管理#

-- 检查脚本是否在缓存中
SCRIPT EXISTS a1b2c3d4... e5f6g7h8...
-- 返回: 1 0 (第一个存在，第二个不存在)

-- 清空所有脚本缓存
SCRIPT FLUSH

-- 清空并同步（Redis 7.0+）
SCRIPT FLUSH SYNC
SCRIPT FLUSH ASYNC

3.4 客户端最佳实践#

# Python: 自动 EVAL → EVALSHA 降级
r = redis.Redis()

# redis-py 内部自动处理:
# 1. 计算 script 的 SHA1
# 2. 尝试 EVALSHA
# 3. 如果 NOSCRIPT 错误 → 降级为 EVAL
script = r.register_script("""
    local stock = tonumber(redis.call('GET', KEYS[1]))
    if stock and stock > 0 then
        redis.call('DECR', KEYS[1])
        return 1
    end
    return 0
""")
result = script(keys=['stock:item1'])

4. Lua 沙箱限制#

4.1 安全限制#

-- 禁止的操作:
os.execute('rm -rf /')   -- 禁止系统调用
io.open('/etc/passwd')    -- 禁止文件操作
require('socket')         -- 禁止加载模块

-- 允许的函数:
redis.call()              -- 调用 Redis 命令
redis.pcall()             -- 调用 Redis 命令（安全模式）
redis.log()               -- 写日志
redis.sha1hex()           -- SHA1 计算
redis.status_reply()      -- 构造状态回复
redis.error_reply()       -- 构造错误回复
cjson.encode()            -- JSON 编码
cjson.decode()            -- JSON 解码
cmsgpack.pack()           -- MessagePack 编码
cmsgpack.unpack()         -- MessagePack 解码

4.2 时间限制#

-- Lua 脚本最大执行时间（毫秒），0=无限制
lua-time-limit 5000

-- 超时后:
-- 1. 其他客户端收到 BUSY 错误
-- 2. 可执行 SCRIPT KILL 终止脚本
-- 3. 如果脚本正在写入，只能 shutdown nosave

4.3 确定性限制（Redis 7.0+）#

Redis 7.0 引入效果复制（effect replication），脚本默认必须确定性：

-- 非确定性脚本（每次执行结果不同）
math.random()              -- 禁止
redis.call('TIME')         -- 禁止
redis.call('SRANDMEMBER')  -- 禁止

-- 如果需要非确定性，使用 redis.set_repl()
redis.set_repl(redis.REPL_ALL)  -- 默认，复制所有写命令
redis.set_repl(redis.REPL_NONE) -- 不复制

5. 实战模式#

5.1 分布式锁释放#

-- 原子性检查并释放锁
if redis.call('GET', KEYS[1]) == ARGV[1] then
    return redis.call('DEL', KEYS[1])
end
return 0

5.2 限流器（滑动窗口）#

local key = KEYS[1]
local limit = tonumber(ARGV[1])
local window = tonumber(ARGV[2])
local now = tonumber(ARGV[3])

redis.call('ZREMRANGEBYSCORE', key, 0, now - window)
local count = redis.call('ZCARD', key)
if count < limit then
    redis.call('ZADD', key, now, now .. '-' .. math.random(1000000))
    redis.call('PEXPIRE', key, window)
    return 1
end
return 0

5.3 库存扣减#

local stock_key = KEYS[1]
local user_key = KEYS[2]
local user_id = ARGV[1]
local quantity = tonumber(ARGV[2])

-- 检查是否已购买
if redis.call('SISMEMBER', user_key, user_id) == 1 then
    return -1  -- 已购买
end

-- 检查库存
local stock = tonumber(redis.call('GET', stock_key))
if not stock or stock < quantity then
    return 0  -- 库存不足
end

-- 扣减库存 + 记录用户
redis.call('DECRBY', stock_key, quantity)
redis.call('SADD', user_key, user_id)
return 1  -- 成功

EVAL 命令#

基本写法：执行 Lua 脚本 EVAL <script> <numkeys> <key> [key ...] <arg> [arg ...]

# 执行简单的 GET + SET 脚本
EVAL "redis.call('SET', KEYS[1], ARGV[1]); return 'OK'" 1 mykey myvalue

基本写法：原子性限流器 EVAL <script> 1 <key> <expire> <limit>

# 使用 Lua 脚本实现原子性限流器
EVAL "local count = redis.call('INCR', KEYS[1]) if count == 1 then redis.call('EXPIRE', KEYS[1], ARGV[1]) end if count > tonumber(ARGV[2]) then return 0 end return 1" 1 rate_limit:user1 60 100

redis.call 与 redis.pcall#

基本写法：redis.call 错误时脚本终止 redis.call(<command>, <args>)

-- redis.call 遇到错误时脚本终止
local val = redis.call('INCR', 'non_numeric_key')

基本写法：redis.pcall 错误时返回错误表 redis.pcall(<command>, <args>)

-- redis.pcall 遇到错误时返回错误表，脚本继续执行
local result = redis.pcall('INCR', 'non_numeric_key')
if type(result) == 'table' and result.err then
    redis.call('SET', 'error_log', result.err)
end

EVALSHA 与脚本缓存#

基本写法：加载脚本到缓存 SCRIPT LOAD <script>

# 加载脚本到缓存，返回 SHA1 校验和
SCRIPT LOAD "return redis.call('GET', KEYS[1])"

基本写法：使用 SHA1 执行缓存脚本 EVALSHA <sha1> <numkeys> <key> [key ...] <arg> [arg ...]

# 使用 SHA1 执行已缓存的脚本
EVALSHA a1b2c3d4e5f6 1 mykey

单脚本写法：检查单个脚本是否在缓存中 SCRIPT EXISTS <sha1>

# 检查单个脚本是否在缓存中
SCRIPT EXISTS a1b2c3d4e5f6

多脚本写法：检查多个脚本是否在缓存中 SCRIPT EXISTS <sha1> [sha1 ...]

# 检查多个脚本是否在缓存中
SCRIPT EXISTS a1b2c3d4e5f6 f7g8h9i0j1k2

基本写法：清空脚本缓存 SCRIPT FLUSH

# 清空所有脚本缓存
SCRIPT FLUSH

基本写法：同步清空脚本缓存 SCRIPT FLUSH SYNC

# 同步方式清空脚本缓存（Redis 7.0+）
SCRIPT FLUSH SYNC

基本写法：异步清空脚本缓存 SCRIPT FLUSH ASYNC

# 异步方式清空脚本缓存（Redis 7.0+）
SCRIPT FLUSH ASYNC

基本写法：Python EVALSHA 自动降级 r.register_script(<script>)

# Python redis-py 自动处理 EVAL 到 EVALSHA 的降级
import redis
r = redis.Redis()

script = r.register_script("""
    local stock = tonumber(redis.call('GET', KEYS[1]))
    if stock and stock > 0 then
        redis.call('DECR', KEYS[1])
        return 1
    end
    return 0
""")
result = script(keys=['stock:item1'])

Lua 沙箱限制#

基本写法：调用 Redis 命令 redis.call(<command>, <args>)

-- 在 Lua 脚本中调用 Redis 命令
redis.call('SET', 'key', 'value')

基本写法：安全模式调用 Redis 命令 redis.pcall(<command>, <args>)

-- 安全模式调用 Redis 命令，错误时不终止脚本
redis.pcall('GET', 'key')

基本写法：写日志 redis.log(<level>, <message>)

-- 在 Lua 脚本中写日志
redis.log(redis.LOG_WARNING, 'something went wrong')

基本写法：SHA1 计算 redis.sha1hex(<string>)

-- 计算字符串的 SHA1 哈希值
local hash = redis.sha1hex('hello')

基本写法：构造状态回复 redis.status_reply(<message>)

-- 构造状态回复
return redis.status_reply('OK')

基本写法：构造错误回复 redis.error_reply(<message>)

-- 构造错误回复
return redis.error_reply('something went wrong')

基本写法：JSON 编码 cjson.encode(<value>)

-- 将 Lua 表编码为 JSON 字符串
local json_str = cjson.encode({name='redis', version=7})

基本写法：JSON 解码 cjson.decode(<json_string>)

-- 将 JSON 字符串解码为 Lua 表
local data = cjson.decode('{"name":"redis","version":7}')

基本写法：MessagePack 编码 cmsgpack.pack(<value>)

-- 将 Lua 表编码为 MessagePack 二进制
local packed = cmsgpack.pack({1, 2, 3})

基本写法：MessagePack 解码 cmsgpack.unpack(<packed_string>)

-- 将 MessagePack 二进制解码为 Lua 表
local data = cmsgpack.unpack(packed_string)

禁止写法：系统调用 os.execute(<command>)

-- 沙箱禁止系统调用
os.execute('rm -rf /')

禁止写法：文件操作 io.open(<path>)

-- 沙箱禁止文件操作
io.open('/etc/passwd')

禁止写法：加载模块 require(<module>)

-- 沙箱禁止加载外部模块
require('socket')

基本写法：设置脚本最大执行时间 lua-time-limit <ms>

# 配置 Lua 脚本最大执行时间为5000毫秒
lua-time-limit 5000

基本写法：复制所有写命令（默认） redis.set_repl(redis.REPL_ALL)

-- 默认行为，复制所有写命令到从节点
redis.set_repl(redis.REPL_ALL)

基本写法：不复制写命令 redis.set_repl(redis.REPL_NONE)

-- 不复制写命令到从节点
redis.set_repl(redis.REPL_NONE)

实战模式#

基本写法：分布式锁释放 EVAL <script> 1 <lock_key> <lock_value>

# 原子性检查并释放分布式锁
EVAL "if redis.call('GET', KEYS[1]) == ARGV[1] then return redis.call('DEL', KEYS[1]) end return 0" 1 lock:resource1 my_token

基本写法：滑动窗口限流器 EVAL <script> 1 <key> <limit> <window> <now>

# 基于 ZSET 实现滑动窗口限流
EVAL "local key = KEYS[1] local limit = tonumber(ARGV[1]) local window = tonumber(ARGV[2]) local now = tonumber(ARGV[3]) redis.call('ZREMRANGEBYSCORE', key, 0, now - window) local count = redis.call('ZCARD', key) if count < limit then redis.call('ZADD', key, now, now .. '-' .. math.random(1000000)) redis.call('PEXPIRE', key, window) return 1 end return 0" 1 rate_limit:user1 100 60000 1718334600000

基本写法：库存扣减 EVAL <script> 2 <stock_key> <user_key> <user_id> <quantity>

# 原子性检查库存并扣减
EVAL "local stock_key = KEYS[1] local user_key = KEYS[2] local user_id = ARGV[1] local quantity = tonumber(ARGV[2]) if redis.call('SISMEMBER', user_key, user_id) == 1 then return -1 end local stock = tonumber(redis.call('GET', stock_key)) if not stock or stock < quantity then return 0 end redis.call('DECRBY', stock_key, quantity) redis.call('SADD', user_key, user_id) return 1" 2 stock:item1 users:item1 user42 1