有勇气的牛排博客
☰

Redis Lua 脚本操作教程

进群口令：博客

1 前言

1.1 为什么要用 Lua 操作 Redis?

Redis本身支持强大的命令操作，但是在高并发场景下肯呢个存在如下问题：

  1. 原子性：多个命令组合操作可能被并发打断，Lua脚本在Redis内部执行，保证单线程原子性。
  2. 减少网络往返：多个命令一次性执行，减少客户端与Redis的多次通信。
  3. 复杂逻辑处理：Lua能写 if-else、循环，封装复杂的业务逻辑。

所以秒杀、库存扣减、分布式锁，都会大量使用到 Lua。

2 Redis执行Lua脚本的方式

2.1 EVAL

执行执行脚本

语法：

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

script：Lua脚本内容

numkeys：传入的 key 数量

key [key ...]：传入的key

arg [arg ...]：额外参数（Lua中通过 ARGV 获取）

2.2 EVALSHA

  • 为了避免脚本过长重复传输，Redis会缓存脚本。
  • 使用 SCRIPT LOAD 加载脚本，返回 sha1。
  • 用 EVAALSHA sha1 numkeys ... 来执行。

3 Lua 脚本在 Redis 中的变量

在Redis的Lua脚本中，所有参数和key都要通过一下两个内置变量传入：

  • KEYS：数组，存放传入的key。
  • ARGV：数组，存放传入的参数。

案例

-- Lua 脚本
local key = KEYS[1]
local value = ARGV[1]
return redis.call("set", key, value)

执行

EVAL "local key=KEYS[1]; local value=ARGV[1]; return redis.call('SET', key, value)" 1 mykey hello

4 Lua 操作 Redis 的常用方法

4.1 redis.call

redis.call("set", "foo", "bar")

如果执行失败会抛出异常。

4.2 redis.pcall

redis.pcall("GET", "foo")

失败不会报错，会返回错误对象。

5 跨语言调用

5.1 python

案例一

redis==6.0.0

# -*- coding: utf-8 -*-
import redis

r = redis.Redis(host='127.0.0.1', port=6379, db=0, decode_responses=True)
print(r)

# 正确的 Lua 脚本：设置 key-value
lua_script = """
return redis.call("set", KEYS[1], ARGV[1])
"""

# 注册脚本
script = r.register_script(lua_script)

# 执行脚本
res = script(keys=['nickname'], args=['有勇气的牛排'])  # nickname=有勇气的牛排
print(res)  # 输出 OK

案例二：异步案例

redis==6.4.0

import asyncio
import redis.asyncio as aioredis


async def main():
    r = aioredis.Redis(host="127.0.0.1", port=6379, db=0, decode_responses=True)

    # 正确的 Lua 脚本：设置 key-value
    lua_script = """
    return redis.call("set", KEYS[1], ARGV[1])
    """

    # 注册脚本
    script = r.register_script(lua_script)

    # 执行脚本
    res = await script(keys=['nickname'], args=['有勇气的牛排'])  # nickname=有勇气的牛排
    print(res)  # 输出 OK

    # 关闭
    await r.aclose()


asyncio.run(main())

Python操作Lua脚本

6 实战案例

6.1 原子性库存扣减(秒杀场景)

输出初始化

product:1      9
product:2      0

LUA

-- 判断库存是否充足，如果大于0则减1，否则返回0
local key = KEYS[1]
local num = tonumber(redis.call("GET", key))

if num > 0 then
    redis.call("DECR", "product:1")
    return 1
else
    return 0
end

Python

import asyncio
import redis.asyncio as aioredis


async def main():
    r = aioredis.Redis(host="127.0.0.1", port=6379, db=0, decode_responses=True)

    # 正确的 Lua 脚本：设置 key-value
    lua_script = """
local key = KEYS[1]
local num = tonumber(redis.call("GET", key))

if num > 0 then
    redis.call("DECR", "product:1")
    return 1
else
    return 0
end
    """

    # 注册脚本
    script = r.register_script(lua_script)

    # 执行脚本
    res = await script(keys=["product:1"], args=[])  # nickname=有勇气的牛排
    print(res)  # 输出 1
    res = await script(keys=["product:2"], args=[])  # nickname=有勇气的牛排
    print(res)  # 输出 0

    # 关闭
    await r.aclose()


asyncio.run(main())

原子性库存扣减(秒杀场景)

6.2 分布式锁

加锁

-- 尝试加锁
-- KEYS[1] 锁的key
-- ARGV[1] 锁的唯一标识（请求ID）
-- ARGV[2] 过期时间
if redis.call("SETNX", KEYS[1], ARGV[1]) == 1 then
    redis.call("PEXPIRE", KEYS[1], ARGV[2])
    return 1
else
    return 0
end

解锁

-- 只允许加锁者自己解锁
if redis.call("GET", KEYS[1]) == ARGV[1] then
    return redis.call("DEL", KEYS[1])
else
    return 0
end

6.3 计数器(限流)

-- 限制某用户每分钟最多访问5次
-- KEYS[1] 用户访问计数Key
-- ARGV[1] 过期时间（秒）
-- ARGV[2] 最大次数
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 -- 超过限制
else
    return 1 -- 允许访问
end

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 else return 1 end" 1 user:123 60 5

7 最佳实践

短小脚本优先：脚本越短越好，复杂逻辑放到应用层。

避免长阻塞：Lua 在 Redis 内部执行，脚本执行期间 Redis 不会处理其他请求，避免死循环。

保证幂等性：脚本最好设计为幂等操作，避免重复执行造成数据错误。

  • 可以给每个操作分配唯一id，判断是否处理过。
  • 常见于：网络重试、MQ多次发送同一条消息、应用崩溃重试。

脚本调试：可以用 redis-cli --ldb --eval 调试 Lua。

评论区

×
×