Python操作Redis数据库的详细教程与应用实战

目录

• 引言
• 安装redis-py库
• 连接到Redis服务器
• 连接到本地Redis服务器
• 连接到远程Redis服务器
• 数据操作
• 键值对操作
• 哈希表操作
• 列表操作
• 集合操作
• 有序集合操作
• 事务
• 订阅与发布
• 管道（Pipeline）
• 持久化与备份
• 总结

pip install redis

import redis

# 使用StrictRedis连接到本地Redis服务器，默认端口为6379
r = redis.StrictRedis(host='localhost', port=6379, db=0, decode_responses=True)

# 或者使用Redis类（二者在大多数情况下可以互换使用）
# r = redis.Redis(host='localhost', port=6379, db=0, decode_responses=True)

# 注意：设置decode_responses=True可以让返回的字符串数据以Python字符串形式存在，而不是默认的字节类型

r = redis.StrictRedis(host='192.168.xxx.xxx', port=16379, password='your_password', db=0, decode_responses=True)

r.set('mykey', 'myvalue')

value = r.get('mykey')
print(value)  # 输出: myvalue

exists = r.exists('mykey')
print(exists)  # 输出: True 或 False

deleted = r.delete('mykey')
print(deleted)  # 输出: 1 表示删除成功

r.mset({'key1': 'value1', 'key2': 'value2', 'key3': 'value3'})

values = r.mget('key1', 'key2', 'key3')
print(values)  # 输出: [b'value1', b'value2', b'value3']

r.hset('myhash', 'field1', 'value1')
r.hset('myhash', 'field2', 'value2')

field_value = r.hget('myhash', 'field1')
print(field_value)  # 输出: value1

deleted_count = r.hdel('myhash', 'field1')
print(deleted_count)  # 输出: 1 表示成功删除一个字段

all_fields_values = r.hgetall('myhash')
print(all_fields_values)  # 输出: {'field2': 'value2'}（注意：实际输出会根据哈希表中的内容而定）

r.rpush('mylist', 'item1', 'item2')

r.lpush('mylist', 'item0')

items = r.lrange('mylist', 0, -1)  # 获取列表中的所有元素，-1 表示列表的最后一个元素
print(items)  # 输出: ['item0', 'item1', 'item2']

first_item = r.lpop('mylist')
print(first_item)  # 输出: item0

last_item = r.rpop('mylist')
print(last_item)  # 输出: item2

r.sadd('myset', 'member1', 'member2')

members = r.smembers('myset')
print(members)  # 输出: {'member1', 'member2'}（注意：集合是无序的，所以每次输出的顺序可能不同）

# 假设有另一个集合
r.sadd('anotherset', 'member1', 'member3')

# 获取两个集合的并集
union_members = r.sunion('myset', 'anotherset')
print(union_members)  # 输出: {'member1', 'member2', 'member3'}

# 获取两个集合的交集
intersect_members = r.sinter('myset', 'anotherset')
print(intersect_members)  # 输出: {'member1'}

# 获取第一个集合与第二个集合的差集
diff_members = r.sdiff('myset', 'anotherset')
print(diff_members)  # 输出: {'member2'}

r.zadd('mysortedset', {'member1': 1, 'member2': 2})

members_with_scores = r.zrange('mysortedset', 0, -1, withscores=True)
print(members_with_scores)  # 输出: [('member1', 1.0), ('member2', 2.0)]

members_with_scores_reversed = r.zrevrange('mysortedset', 0, -1, withscores=True)
print(members_with_scores_reversed)  # 输出: [('member2', 2.0), ('member1', 1.0)]

removed_count = r.zrem('mysortedset', 'member1')
print(removed_count)  # 输出: 1 表示成功移除了一个元素

pipe = r.pipeline()
try:
    pipe.multi()
    pipe.set('key1', 'value1')
    pipe.set('key2', 'value2')
    pipe.execute()  # 执行事务
except redis.exceptions.RedisError as e:
    print(f"事务执行失败: {e}")
    pipe.reset()  # 取消事务中的所有命令

r.publish('mychannel', 'Hello, Redis!')

import redis
import threading

def listener(r, channel):
    pubsub = r.pubsub()
    pubsub.subscribe(channel)
    for message in pubsub.listen():
        if message['type'] == 'message':
            print(f"Received {message['data']} on {message['channel']}")

# 创建一个Redis连接
r = redis.Redis(host='localhost', port=6379, db=0, decode_responses=True)

# 创建一个线程来订阅消息
thread = threading.Thread(target=listener, args=(r, 'mychannel'))
thread.daemon = True
thread.start()

# 在主线程中发布消息
r.publish('mychannel', 'Hello from the main thread!')

# 注意：由于订阅操作是阻塞的，上面的代码在发布消息后不会立即退出。
# 在实际应用中，你可能需要一种方式来优雅地关闭监听线程。

pipe = r.pipeline()
pipe.set('key1', 'value1')
pipe.set('key2', 'value2')
pipe.execute()  # 批量执行命令