Python實現(xiàn)操作Redis的高級用法分享

更新時間：2023年05月21日 11:07:25 作者：CodeDevMaster

redis-py是Python操作Redis的第三方庫，它提供了與Redis服務(wù)器交互的API，本文為大家介紹了Python利用redis-py操作Redis的高級用法，需要的可以收藏一下

redis-py

redis-py是Python操作Redis的第三方庫，它提供了與Redis服務(wù)器交互的API。

GitHub地址：https://github.com/redis/redis-py

安裝redis-py

pip install redis

基本使用

import redis
# 連接到本地Redis服務(wù)器，默認端口為6379
r = redis.Redis(host='localhost', port=6379, db=0)
# 設(shè)置鍵值對
r.set('hello', 'world')
# 獲取鍵對應(yīng)的值
value = r.get('hello')
# 輸出 b'world'
print(value)
# 批量設(shè)置鍵值對
r.mset({'foo': '1', 'bar': '2'})
# 批量獲取鍵對應(yīng)的值
values = r.mget(['foo', 'bar'])
# 輸出 [b'1', b'2']
print(values)

也可以使用StrictRedis對象連接redis服務(wù)器，StrictRedis類基于Redis類實現(xiàn)。

import redis
# 連接到本地Redis服務(wù)器，默認端口為6379
r = redis.StrictRedis(host='localhost', port=6379, db=0)
# 設(shè)置鍵值對
r.set('hello', 'world')
# 獲取鍵對應(yīng)的值
value = r.get('hello')
print(value) # 輸出 b'world'

增刪改查

以操作String數(shù)據(jù)類型的增刪改查為例。

# 引?模塊
import redis
if __name__ == '__main__':
    try:
        # 創(chuàng)建Redis對象
        r = redis.Redis(host='localhost', port=6379, db=0)
        # 新增，添加成功則返回True，如果添加失敗則返回False
        result = r.set('name', 'test')
        print('是否新增成功：', result)
        # 獲取，如果鍵存在則返回對應(yīng)的值，如果鍵不存在則返回None
        name = r.get('name')
        print('查詢結(jié)果：', name)
        # 修改，如果鍵已經(jīng)存在則進?修改，如果鍵不存在則進?添加
        result = r.set('name', 'redis')
        print('是否修改成功：', result)
        name = r.get('name')
        print('查詢結(jié)果：', name)
		# 獲取所有的鍵
        result = r.keys()
        print('獲取所有的鍵', result)
        # 刪除，刪除鍵及對應(yīng)的值，如果刪除成功則返回受影響的鍵數(shù)，否則則返 回0
        result = r.delete('name')
        print('刪除key的數(shù)量：', result)
    except Exception as e:
        print(e)

字符串操作

import redis
# 連接到本地Redis服務(wù)器，默認端口為6379
r = redis.StrictRedis(host='localhost', port=6379, db=0)
# 設(shè)置鍵值對，鍵為'foo'，值為'bar'
r.set('foo', 'bar')
# 獲取鍵對應(yīng)的值
value = r.get('foo')
print(value) # 輸出 b'bar'
# 批量設(shè)置鍵值對
r.mset({'apple': 'red', 'banana': 'yellow'})
# 批量獲取鍵對應(yīng)的值
values = r.mget(['apple', 'banana'])
print(values) # 輸出 [b'red', b'yellow']
# 獲取部分值
part_value = r.getrange('foo', 0, 1)
print(part_value) # 輸出 b'ba'
# 追加字符串
r.append('foo', 'baz')
append_value = r.get('foo')
print(append_value) # 輸出 b'barbaz'
# 自增計數(shù)器
r.incr('counter')
# 獲取計數(shù)器的值
value = r.get('counter')
print(value) # 輸出 b'1'
# 在自增計數(shù)器的基礎(chǔ)上再加上5
r.incrby('counter', 5)
# 獲取計數(shù)器的值
value = r.get('counter')
print(value) # 輸出 b'6'
# 減少計數(shù)器
r.decr('counter')
# 獲取計數(shù)器的值
value = r.get('counter')
print(value) # 輸出 b'5'

哈希操作

import redis
# 連接到本地Redis服務(wù)器，默認端口為6379
r = redis.StrictRedis(host='localhost', port=6379, db=0)
# 設(shè)置哈希表鍵值對
r.hset('user1', 'name', 'Alice')
r.hset('user1', 'age', 20)
r.hset('user1', 'gender', 'female')
# 獲取整個哈希表
hash_table = r.hgetall('user1')
print(hash_table) # 輸出 {b'name': b'Alice', b'age': b'20', b'gender': b'female'}
# 獲取特定鍵對應(yīng)的值
value = r.hget('user1', 'name')
print(value) # 輸出 b'Alice'
# 刪除哈希表的一個鍵值對
r.hdel('user1', 'gender')
# 獲取所有鍵名
keys = r.hkeys('user1')
print(keys) # 輸出 [b'name', b'age']
# 獲取所有鍵名對應(yīng)的值
values = r.hvals('user1')
print(values) # 輸出 [b'Alice', b'20']
# 批量設(shè)置哈希表鍵值對
r.hmset('user2', {'name': 'Bob', 'age': 25})
# 批量獲取哈希表鍵名對應(yīng)的值
values = r.hmget('user2', ['name', 'age'])
print(values) # 輸出 [b'Bob', b'25']

列表操作

import redis
# 連接到本地Redis服務(wù)器，默認端口為6379
r = redis.StrictRedis(host='localhost', port=6379, db=0)
# 從左側(cè)插入元素
r.lpush('mylist1', 'foo')
r.lpush('mylist1', 'bar')
r.lpush('mylist1', 'baz')
# 從右側(cè)刪除元素
r.rpop('mylist1')
# 獲取列表長度
length = r.llen('mylist1')
print(length) # 輸出 2
# 獲取整個列表
mylist = r.lrange('mylist1', 0, -1)
print(mylist) # 輸出 [b'baz', b'bar']
# 從左側(cè)插入元素
r.lpush('mylist2', 'one')
r.lpush('mylist2', 'two')
r.lpush('mylist2', 'three')
# 彈出列表頭部元素
value1 = r.lpop('mylist2')
print(value1) # 輸出 b'three'
# 彈出列表尾部元素
value2 = r.rpop('mylist2')
print(value2) # 輸出 b'one'
# 在指定元素前或后插入新元素
r.linsert('mylist2', 'BEFORE', 'two', 'new')
mylist = r.lrange('mylist2', 0, -1)
print(mylist) # 輸出 [b'new', b'two']
# 裁剪列表
r.ltrim('mylist2', 0, 0)
mylist = r.lrange('mylist2', 0, -1)
print(mylist) # 輸出 [b'new']

集合操作

import redis
# 連接到本地Redis服務(wù)器，默認端口為6379
r = redis.StrictRedis(host='localhost', port=6379, db=0)
# 向集合中添加元素
r.sadd('set1', 'foo')
r.sadd('set1', 'bar')
# 獲取集合中的所有元素
members = r.smembers('set1')
print(members) # 輸出 {b'foo', b'bar'}
# 獲取集合中的元素數(shù)量
count = r.scard('set1')
print(count) # 輸出 2
# 判斷一個元素是否在集合中
result = r.sismember('set1', 'foo')
print(result) # 輸出 True
# 刪除集合中的一個元素
r.srem('set1', 'bar')
# 獲取多個集合的交集
r.sadd('set2', 'foo')
r.sadd('set2', 'baz')
intersection = r.sinter(['set1', 'set2'])
print(intersection) # 輸出 {b'foo'}
# 獲取多個集合的并集
union = r.sunion(['set1', 'set2'])
print(union) # 輸出 {b'foo', b'baz'}
# 獲取一個集合與多個集合的差集
difference = r.sdiff('set2', ['set1'])
print(difference) # 輸出 {b'baz'}

有序集合操作

import redis
# 連接到本地Redis服務(wù)器，默認端口為6379
r = redis.StrictRedis(host='localhost', port=6379, db=0)
# 添加有序集合成員和分值
r.zadd('zset1', {'foo': 1.0, 'bar': 2.0, 'baz': 4.0})
# 獲取有序集合的成員數(shù)
count = r.zcard('zset1')
print(count) # 輸出 3
# 獲取有序集合指定范圍內(nèi)的成員
members = r.zrange('zset1', 0, -1)
print(members) # 輸出 [b'foo', b'bar', b'baz']
# 獲取有序集合指定成員的分值
score = r.zscore('zset1', 'bar')
print(score) # 輸出 2.0
# 獲取有序集合指定范圍內(nèi)成員的數(shù)量
count = r.zcount('zset1', 1.5, 3.5)
print(count) # 輸出 1
# 刪除有序集合中一個成員
r.zrem('zset1', 'bar')
# 獲取有序集合中指定范圍內(nèi)的成員和分值
with_scores = r.zrangebyscore('zset1', 0, 5, withscores=True)
print(with_scores) # 輸出 [(b'foo', 1.0), (b'baz', 4.0)]

高級用法

Redis管道pipeline

在Redis中，管道（pipeline）是指可以將多個Redis命令依次發(fā)送給Redis，讓Redis 一次性執(zhí)行這些命令并返回結(jié)果的機制。使用管道可以大大減少客戶端與Redis的網(wǎng)絡(luò)通信次數(shù)，提高Redis的處理效率，是優(yōu)化Redis性能的重要手段之一。

在redis-py庫中，可以使用pipeline()方法創(chuàng)建一個管道對象，并對該對象連續(xù)調(diào)用多個 Redis 命令并提交到 Redis 進行執(zhí)行。提交執(zhí)行后，每個命令都會獲取到這些命令的執(zhí)行結(jié)果，并按照請求的順序返回給客戶端。

特點：

1.可以一次性發(fā)送多條命令并在執(zhí)行完后一次性將結(jié)果返回

2.pipeline通過減少客戶端與Redis的通信次數(shù)來實現(xiàn)降低往返延時時間

實現(xiàn)原理：

管道pipeline實現(xiàn)的原理是隊列，隊列是先進先出，這樣就保證數(shù)據(jù)的順序性
Client可以將三個命令放到一個tcp報文一起發(fā)送
Server則可以將三條命令的處理結(jié)果放到一個tcp報文返回

基本使用

1.使用 pipeline() 方法創(chuàng)建一個新的 Pipeline 對象，并向該管道對象連續(xù)調(diào)用了三個不同的 SET 命令，分別設(shè)置了三個不同的鍵名和對應(yīng)的鍵值

2.通過 execute() 方法提交管道內(nèi)所有的命令。Redis 服務(wù)器一次性執(zhí)行管道內(nèi)所有的命令，并將結(jié)果返回給客戶端

3.最后輸出Redis 管道執(zhí)行的結(jié)果到控制臺，其中包含了每個 SET 命令的執(zhí)行結(jié)果

import redis
# 連接到本地 Redis 服務(wù)器，默認端口為 6379
r = redis.StrictRedis(host='localhost', port=6379, db=0)
# 使用 Redis 管道
pipe = r.pipeline()
# 連續(xù)設(shè)置多個鍵名和鍵值
pipe.set('name', 'John')
pipe.set('age', 30)
pipe.set('city', 'New York')
# 執(zhí)行 Redis 管道中所有命令，并獲取所有命令的執(zhí)行結(jié)果
result = pipe.execute()
# 輸出 Redis 管道執(zhí)行結(jié)果
print(result)

Redis事務(wù)

事務(wù) Redis 通過 MULTI 和 EXEC 來實現(xiàn)事務(wù)，MULTI 開啟一個事務(wù)，EXEC 提交多個命令到 Redis 執(zhí)行，可以保證單位時間內(nèi)只有當(dāng)前請求在訪問 Redis 服務(wù)器，其他讀寫操作會等待這個事務(wù)結(jié)束后才能進行，從而保證了數(shù)據(jù)一致性。

基本使用

1.創(chuàng)建一個 redis.StrictRedis 實例，并且設(shè)置 transaction 參數(shù)為 True，表示開啟 Redis 事務(wù)

2.使用 pipeline() 方法創(chuàng)建一個新的 Pipeline 對象，并將其 transaction 參數(shù)設(shè)置為 True，表示這個 Pipeline 是用于 Redis 事務(wù)的

3.調(diào)用 multi() 方法開啟Redis 事務(wù)，之后向兩個不同的鍵名 foo 和 bar 分別設(shè)置了不同的字符串值

4.最后通過 execute() 方法提交事務(wù)，Redis 將一次性執(zhí)行整個事務(wù)并返回每個命令的執(zhí)行結(jié)果

import redis
# 連接到本地 Redis 服務(wù)器，默認端口為 6379
r = redis.StrictRedis(host='localhost', port=6379, db=0)
# 使用 Redis 事務(wù)
trans = r.pipeline(transaction=True)
# 開啟事務(wù)
trans.multi()
# 向兩個鍵名分別設(shè)置不同的值
trans.set('foo', 'hello')
trans.set('bar', 'world')
# 在 Redis 事務(wù)中執(zhí)行以上命令，并獲取執(zhí)行結(jié)果
result = trans.execute()
# 輸出 Redis 事務(wù)執(zhí)行結(jié)果
print(result)

事務(wù)與管道的區(qū)別

在Redis中，事務(wù)（transaction）和管道（pipeline）都是用于批量執(zhí)行命令的方式，但二者有本質(zhì)上的不同：

1.調(diào)用方式不同

使用事務(wù)時，需要先通過MULTI命令將客戶端設(shè)置為事務(wù)模式，然后按照一定的順序添加執(zhí)行的多個命令，最后通過EXEC命令將操作提交到服務(wù)器執(zhí)行。

使用管道時，則是對同一個連接對象上連續(xù)調(diào)用多個Redis命令并且在最后統(tǒng)一執(zhí)行這些命令。

2.發(fā)送機制不同

Redis事務(wù)的邏輯單元可以確保所有被包含的命令“原子性”地執(zhí)行，即要么全部執(zhí)行成功完成，要么全部回滾；而Redis>運用管道的方法僅僅是優(yōu)化傳輸，將多個命令打包發(fā)送到Redis服務(wù)節(jié)點，并在結(jié)果關(guān)閉時進行收集處理，以達到多個請求一次通信的目的。

3.回滾能力不同

Redis事務(wù)提交的過程中如果某個命令執(zhí)行失敗了，后面的命令則都不會再執(zhí)行，已經(jīng)執(zhí)行過的命令不會回滾。當(dāng)然在EXEC之前可以通過DISCARD命令清空已經(jīng)放入到事務(wù)隊列里面的命令；而管道機制暫時沒有回滾的能力。

因此:

Redis管道是解決高性能I/O操作的手段，主要目的在于將多個命令打包，一次發(fā)出去避免了每次發(fā)送都占有一個網(wǎng)絡(luò)通道

Redis事務(wù)適用于數(shù)據(jù)的批量修改，并期望原子性action。兩種方案各有利弊，需要按照實際業(yè)務(wù)場景選擇使用哪一種方式。

分布式鎖

Redis通過 SETNX 和 EXPIRE 等命令實現(xiàn)分布式鎖，可防止多個客戶端同時修改同一資源。具體實現(xiàn)時，檢查一個鍵是否存在，若不存在則對該鍵進行設(shè)置并獲得鎖；若已存在則等待。

import time
import uuid
import redis
# 連接到本地 Redis 服務(wù)器，默認端口為 6379
r = redis.StrictRedis(host='localhost', port=6379, db=0)
def acquire_lock(conn, lockname, acquire_timeout=10, lock_timeout=10):
    """
    嘗試獲取鎖，獲取成功返回鎖id；獲取失敗或者異常返回None
    :param conn: Redis連接對象
    :param lockname: 鎖的名稱
    :param acquire_timeout: 最大嘗試獲取鎖的時間（seconds）
    :param lock_timeout: 鎖的超時時間（seconds）
    :return 是否獲取權(quán)益：鎖的ID/None
    """
    identifier = str(uuid.uuid4())
    end_time = time.time() + acquire_timeout
    while time.time() < end_time:
        # 獲取鎖
        if conn.setnx(lockname, identifier):
            conn.expire(lockname, lock_timeout)
            return identifier
        # 防止死鎖
        elif not conn.ttl(lockname):
            conn.expire(lockname, lock_timeout)
        time.sleep(0.001)
    return None
def release_lock(conn, lockname, identifier):
    """
    根據(jù)鎖id釋放鎖，若鎖不存在或者已經(jīng)被其他持有者所釋放則返回False；成功釋放返回True
    :param conn: Redis連接對象
    :param lockname: 鎖的名稱
    :param identifier：鎖的ID
    :return 是否釋放：True/False
    """
    pipe = conn.pipeline(True)
    while True:
        try:
            # 開啟事務(wù)
            pipe.watch(lockname)
            # 檢查對應(yīng)鍵是否還是要當(dāng)前程序設(shè)置的值，以avoid誤刪別的客戶端的鎖
            if pipe.get(lockname).decode('utf-8') == identifier:
                # 刪除鎖
                pipe.multi()
                pipe.delete(lockname)
                pipe.execute()
                return True
            pipe.unwatch()
            break
        except redis.exceptions.WatchError:
            pass
    return False
if __name__ == '__main__':
    lockname = " lock"
    id = acquire_lock(r, lockname)
    print(id)
    tag = release_lock(r, lockname, id)
    print(tag)

訂閱和發(fā)布

Redis 通過 SUBSCRIBE 和 PUBLISH 命令實現(xiàn)類似消息隊列的功能，可以實現(xiàn)多個進程（客戶端）針對一個頻道進行消息的監(jiān)聽和廣播。

import redis
# 連接到本地 Redis 服務(wù)器，默認端口為 6379
r = redis.StrictRedis(host='localhost', port=6379, db=0)
# 在 “頻道” 上發(fā)布 “消息”
pub = r.publish('channel', 'Hello World')
# 返回發(fā)布/訂閱對象
sub = r.pubsub()
# 使用此對象，可以訂閱頻道并收聽發(fā)布的消息
sub.subscribe('channel')
for message in sub.listen():
    print(message)
    print(message['data'])

GeoHash

Redis通過GeoHash 實現(xiàn)了地理位置排序和搜索等功能。

import redis
# 連接到本地Redis服務(wù)器，默認端口為6379
r = redis.StrictRedis(host='localhost', port=6379, db=0)
# 添加地理位置坐標
r.geoadd("locations", 126.6507114792, 45.7603340486, "Harbin")
r.geoadd("locations", 117.2028132333, 39.0879960175, "Beijing")
r.geoadd("locations", 121.5670484719, 38.9544883509, "Tianjin")
# 獲取指定兩點間的距離（以 km 為單位）
distance = r.geodist("locations", "Harbin", "Beijing", unit="km")
print(distance)  # 1072.1429
# 搜索指定范圍內(nèi)的地理位置（以 km 為單位），并按照距離從近到遠排序
locations = r.georadiusbymember("locations", "Tianjin", 500, unit="km", withdist=True, sort="ASC")
print(locations)  # [[b'Tianjin', 0.0], [b'Beijing', 377.3833]]

redis-py-cluster

redis-py-cluster是Python中用于連接Redis集群的模塊，支持對Redis集群中的所有節(jié)點進行Hash槽分配操作，提供了與redis-py相同的API接口，使用方法類似。

GitHub地址：https://github.com/Grokzen/redis-py-cluster

安裝

pip install redis-py-cluster

基本使用

from rediscluster import RedisCluster
# Redis節(jié)點
startup_nodes = [
    {"host": "127.0.0.1", "port": "6379"},
    {"host": "127.0.0.1", "port": "7001"},
    {"host": "127.0.0.1", "port": "7002"}
]
# 創(chuàng)建RedisCluster對象
rc = RedisCluster(startup_nodes=startup_nodes, decode_responses=True)
# 與Redis的基本交互
rc.set("foo", "bar")
value = rc.get("foo")
print(value)  # bar
# 哈希操作
rc.hset("user", "name", "Alice")
rc.hset("user", "age", 20)
hash_table = rc.hgetall("user")
print(hash_table)  # {'name': 'Alice', 'age': '20'}
# 列表操作
rc.lpush("mylist", "foo")
rc.lpush("mylist", "bar")
rc.lpush("mylist", "baz")
mylist = rc.lrange("mylist", 0, -1)
print(mylist)  # ['baz', 'bar', 'foo']
# 刪除鍵
rc.delete("foo")

以上就是Python實現(xiàn)操作Redis的高級用法分享的詳細內(nèi)容，更多關(guān)于Python操作Redis的資料請關(guān)注腳本之家其它相關(guān)文章！

您可能感興趣的文章: