方法一：使用 pipeline

使用pipelining 發(fā)送命令時(shí)，redis server必須部分請(qǐng)求放到隊(duì)列中（使用內(nèi)存）執(zhí)行完畢后一次性發(fā)送結(jié)果，在 pipeline 使用期間，將“獨(dú)占”鏈接，無(wú)法進(jìn)行非“管道”類(lèi)型的其他操作，直至 pipeline 關(guān)閉；如果 pipeline 的指令集很多很龐大，為了不影響其他操作（redis 最大時(shí)間lua-time-limit默認(rèn)是5s），可以使用其他新建新鏈接操作。批量操作如下：

import redisr = 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 以樂(lè)觀鎖的形式執(zhí)行事務(wù)操作

# -*- coding:utf-8 -*-import redisfrom redis import WatchErrorfrom concurrent.futures import ProcessPoolExecutorr = redis.Redis(host=’127.0.0.1’, port=6379)# 減庫(kù)存函數(shù), 循環(huán)直到減庫(kù)存完成# 庫(kù)存充足, 減庫(kù)存成功, 返回True# 庫(kù)存不足, 減庫(kù)存失敗, 返回Falsedef decr_stock(): # python中redis事務(wù)是通過(guò)pipeline的封裝實(shí)現(xiàn)的 with r.pipeline() as pipe: while True: try:# watch庫(kù)存鍵, multi后如果該key被其他客戶(hù)端改變, 事務(wù)操作會(huì)拋出WatchError異常pipe.watch(’stock:count’)count = int(pipe.get(’stock:count’))if count > 0: # 有庫(kù)存 # 事務(wù)開(kāi)始 pipe.multi() # multi 判斷 watch 監(jiān)控的 key 是否被其他客戶(hù)端改變 pipe.decr(’stock:count’) # 把命令推送過(guò)去 # execute返回命令執(zhí)行結(jié)果列表, 這里只有一個(gè)decr返回當(dāng)前值 result = pipe.execute()[0] print('result: ', result) return Trueelse: return False except WatchError as e:# 打印WatchError異常, 觀察被watch鎖住的情況print(e.args) finally:pipe.unwatch()def worker(): while True: # 沒(méi)有庫(kù)存就退出 if not decr_stock(): break# 實(shí)驗(yàn)開(kāi)始# 設(shè)置庫(kù)存為100r.set('stock:count', 100)# 多進(jìn)程模擬多個(gè)客戶(hù)端提交with ProcessPoolExecutor(max_workers=2) as pool: for _ in range(10): pool.submit(worker)方法二：使用 register_script

分布執(zhí)行，發(fā)送腳本到redis服務(wù)器，獲取一個(gè)本次連接的一個(gè)調(diào)用句柄，根據(jù)此句柄可以無(wú)數(shù)次執(zhí)行不同參數(shù)調(diào)用

import redisimport 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 來(lái)實(shí)現(xiàn)，需要注意 redis.call(method, key, field) 的返回值（nil，false，1），此處沒(méi)有鍵值返回的是false。如果中間有錯(cuò)誤，所有的語(yǔ)句不時(shí)不生效。

方法三：使用 script_load 和 evalsha

簡(jiǎn)而言之，通過(guò) script_load 發(fā)送給redis服務(wù)器，使加載 lua 腳本，并常駐內(nèi)存，返回標(biāo)志，通過(guò) evalsha 按標(biāo)志進(jìn)行執(zhí)行，此連接脫離本次redis 客戶(hù)端。

import redisimport 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_... ）

script load

此命令用于將Lua腳本加載到Redis內(nèi)存中

script exists

scripts exists sha1 [sha1 …] 此命令用于判斷sha1是否已經(jīng)加載到Redis內(nèi)存中

script flush

此命令用于清除Redis內(nèi)存已經(jīng)加載的所有Lua腳本，在執(zhí)行script flush后，所有 sha 不復(fù)存在。

script kill

此命令用于殺掉正在執(zhí)行的Lua腳本。

方法四：eval

使用方法與方法三類(lèi)似，但是eval是一次性請(qǐng)求，每次請(qǐng)求，必須攜帶 lua 腳本

以上就是用python 批量操作redis數(shù)據(jù)庫(kù)的詳細(xì)內(nèi)容，更多關(guān)于python 批量操作redis數(shù)據(jù)庫(kù)的資料請(qǐng)關(guān)注好吧啦網(wǎng)其它相關(guān)文章！