Python Redis如何執(zhí)行Lua腳本

更新時(shí)間：2024年02月09日 08:38:07 作者：XerCis CSDN認(rèn)證博客專(zhuān)家

這篇文章主要介紹了Python Redis如何執(zhí)行Lua腳本問(wèn)題,具有很好的參考價(jià)值,希望對(duì)大家有所幫助,如有錯(cuò)誤或未考慮完全的地方,望不吝賜教

問(wèn)題描述

Python Redis 執(zhí)行 Lua 腳本可實(shí)現(xiàn)原子性

官方文檔表示，Redis 使用相同的 Lua 解釋器來(lái)運(yùn)行所有命令，Redis 保證腳本以原子方式執(zhí)行：

在執(zhí)行腳本時(shí)不會(huì)執(zhí)行其他腳本或 Redis 命令。
這種語(yǔ)義類(lèi)似于 MULTI / EXEC 的語(yǔ)義，從所有其他客戶(hù)的角度來(lái)看，腳本的效果要么仍然不可見(jiàn)，要么已經(jīng)完成。

解決方案

安裝

pip install redis

代碼

from redis import Redis

redis = Redis()
script = """
redis.call('SETEX', 'key', 100, 'value')
"""
redis.register_script(script)()  # 執(zhí)行腳本

效果

傳遞參數(shù)

Script 類(lèi)調(diào)用函數(shù)原型：

def __call__(self, keys=[], args=[], client=None)

from redis import Redis

redis = Redis()
script = """
local key = KEYS[1]
local seconds = ARGV[1]
local value = ARGV[2]
redis.call('SETEX', key, seconds, value)
"""
cmd = redis.register_script(script)
cmd(keys=['key'], args=[100, 'value'])  # 執(zhí)行腳本

效果同上

分布式鎖

1.加鎖并設(shè)過(guò)期時(shí)間 SET lock_resource_name $uuid NX PX $expire_time，同時(shí)啟動(dòng)守護(hù)線程為快要過(guò)期單還沒(méi)執(zhí)行完畢的客戶(hù)端的鎖續(xù)命;

2.客戶(hù)端執(zhí)行業(yè)務(wù)邏輯操作共享資源；

3.通過(guò) Lua 腳本釋放鎖，先 get 判斷鎖是否是自己加的，再執(zhí)行 DEL。

判斷+刪除

import time
from uuid import uuid4

from redis import Redis

redis = Redis()

# 判斷是否自己創(chuàng)建的鎖，是的話則釋放
script = """
if redis.call('GET', KEYS[1]) == ARGV[1] then
    return redis.call('DEL', KEYS[1])
else
    return 0
end
"""


def do_something():
    name = 'lock'  # 鎖名，此處故意寫(xiě)死讓別的進(jìn)程進(jìn)來(lái)占用
    uuid = str(uuid4())
    # 上鎖
    if redis.set(name=name, value=uuid, nx=True, ex=25):  # 鎖不存在才能上鎖成功，過(guò)期時(shí)間應(yīng)為業(yè)務(wù)時(shí)間的五倍，此處故意寫(xiě)小，模擬過(guò)期釋放
        print('上鎖')
        try:
            print('處理業(yè)務(wù)', uuid)
            time.sleep(10)
        except Exception as e:
            print(e)
        finally:
            # 釋放鎖
            cmd = redis.register_script(script)
            res = cmd(keys=[name], args=[uuid])  # 執(zhí)行腳本
            if res:
                print('鎖已釋放')
            else:
                print('不是自己的鎖')
    else:
        print('鎖已存在')


if __name__ == '__main__':
    do_something()

模擬過(guò)程：