方法一：使用 pipeline
使用pipelining 發(fā)送命令時，redis server必須部分請求放到隊(duì)列中（使用內(nèi)存）執(zhí)行完畢后一次性發(fā)送結(jié)果，在 pipeline 使用期間，將“獨(dú)占”鏈接，無法進(jìn)行非“管道”類型的其他操作，直至 pipeline 關(guān)閉；如果 pipeline 的指令集很多很龐大，為了不影響其他操作（redis 最大時間lua-time-limit默認(rèn)是5s），可以使用其他新建新鏈接操作。批量操作如下：
import redis
r = redis.Redis(host='127.0.0.1', port=6379, password='1234567890')
with r.pipeline() as ctx:
a = time.time()
ctx.hset('current', "time2", a)
ctx.hset('current', "time3", a)
res = ctx.execute()
print("result: ", res)
使用 pipe line 以樂觀鎖的形式執(zhí)行事務(wù)操作
# -*- coding:utf-8 -*-
import redis
from redis import WatchError
from concurrent.futures import ProcessPoolExecutor
r = redis.Redis(host='127.0.0.1', port=6379)
# 減庫存函數(shù), 循環(huán)直到減庫存完成
# 庫存充足, 減庫存成功, 返回True
# 庫存不足, 減庫存失敗, 返回False
def decr_stock():
# python中redis事務(wù)是通過pipeline的封裝實(shí)現(xiàn)的
with r.pipeline() as pipe:
while True:
try:
# watch庫存鍵, multi后如果該key被其他客戶端改變, 事務(wù)操作會拋出WatchError異常
pipe.watch('stock:count')
count = int(pipe.get('stock:count'))
if count > 0: # 有庫存
# 事務(wù)開始
pipe.multi() # multi 判斷 watch 監(jiān)控的 key 是否被其他客戶端改變
pipe.decr('stock:count')
# 把命令推送過去
# execute返回命令執(zhí)行結(jié)果列表, 這里只有一個decr返回當(dāng)前值
result = pipe.execute()[0]
print("result: ", result)
return True
else:
return False
except WatchError as e:
# 打印WatchError異常, 觀察被watch鎖住的情況
print(e.args)
finally:
pipe.unwatch()
def worker():
while True:
# 沒有庫存就退出
if not decr_stock():
break
# 實(shí)驗(yàn)開始
# 設(shè)置庫存為100
r.set("stock:count", 100)
# 多進(jìn)程模擬多個客戶端提交
with ProcessPoolExecutor(max_workers=2) as pool:
for _ in range(10):
pool.submit(worker)
方法二：使用 register_script
分布執(zhí)行，發(fā)送腳本到redis服務(wù)器，獲取一個本次連接的一個調(diào)用句柄，根據(jù)此句柄可以無數(shù)次執(zhí)行不同參數(shù)調(diào)用
import redis
import time
r = redis.Redis(host='127.0.0.1', port=31320, password='12345678')
lua = """
local key = KEYS[1]
local field = ARGV[1]
local timestamp_new = ARGV[2]
-- get timestamp of the key in redis
local timestamp_old = redis.call('hget', key, field)
-- if timestamp_old == nil, it means the key is not exist
if timestamp_old == nil or timestamp_old == false or timestamp_new > timestamp_old then
redis.call('hset', key, field .. 1, timestamp_new)
-- timestamp_new > timestamp_old
return redis.pcall('hset', key, field, timestamp_new)
end
"""
cmd = r.register_script(lua)
cur_time = time.time()
cmd(keys=['current'], args=["time", cur_time])
register_script 調(diào)用 lua 來實(shí)現(xiàn)，需要注意 redis.call(method, key, field) 的返回值（nil，false，1），此處沒有鍵值返回的是false。如果中間有錯誤，所有的語句不時不生效。
方法三：使用 script_load 和 evalsha
簡而言之，通過 script_load 發(fā)送給redis服務(wù)器，使加載 lua 腳本，并常駐內(nèi)存，返回標(biāo)志，通過 evalsha 按標(biāo)志進(jìn)行執(zhí)行，此連接脫離本次redis 客戶端。
import redis
import time
r = redis.Redis(host='127.0.0.1', port=31320, password='12345678')
lua = """
local key = KEYS[1]
local field = ARGV[1]
local timestamp_new = ARGV[2]
-- get timestamp of the key in redis
local timestamp_old = redis.call('hget', key, field)
-- if timestamp_old == nil, it means the key is not exist
if timestamp_old == nil or timestamp_old == false or timestamp_new > timestamp_old then
redis.call('hset', key, field .. 1, timestamp_new)
-- timestamp_new > timestamp_old
return redis.pcall('hset', key, field, timestamp_new)
end
"""
sha = r.script_load(lua)
print(r.evalsha(sha, 1, 'current', 'time', time.time()))
Redis 管理Lua腳本：（Python下為 script_... ）
此命令用于將Lua腳本加載到Redis內(nèi)存中
scripts exists sha1 [sha1 …]
此命令用于判斷sha1是否已經(jīng)加載到Redis內(nèi)存中
此命令用于清除Redis內(nèi)存已經(jīng)加載的所有Lua腳本，在執(zhí)行script flush后，所有 sha 不復(fù)存在。
此命令用于殺掉正在執(zhí)行的Lua腳本。
方法四：eval
使用方法與方法三類似，但是eval是一次性請求，每次請求，必須攜帶 lua 腳本
以上就是用python 批量操作redis數(shù)據(jù)庫的詳細(xì)內(nèi)容，更多關(guān)于python 批量操作redis數(shù)據(jù)庫的資料請關(guān)注腳本之家其它相關(guān)文章！
您可能感興趣的文章:- Python3爬蟲RedisDump的安裝步驟
- Redis分布式鎖python-redis-lock使用方法
- Python調(diào)用Redis的示例代碼
- Python安裝并操作redis實(shí)現(xiàn)流程詳解
- python操作redis數(shù)據(jù)庫的三種方法
- 基于python實(shí)現(xiàn)操作redis及消息隊(duì)列
- python3實(shí)現(xiàn)將json對象存入Redis以及數(shù)據(jù)的導(dǎo)入導(dǎo)出
- Python定時從Mysql提取數(shù)據(jù)存入Redis的實(shí)現(xiàn)
- Python+redis通過限流保護(hù)高并發(fā)系統(tǒng)
- python如何基于redis實(shí)現(xiàn)ip代理池
