从零开始：Python操作Redis完全指南

引言：Redis的魅力与Python的结合

在当今高速发展的数据世界中，对数据处理的实时性、高并发性、低延迟性需求日益增长。传统的关系型数据库在某些场景下，由于其磁盘I/O的限制，难以满足这些苛刻的要求。此时，内存数据库的崛起为我们带来了新的解决方案，而Redis（Remote Dictionary Server）无疑是其中的佼佼者。

Redis是一款开源的、基于内存的、高性能的键值存储系统，它不仅仅是一个简单的键值数据库，更是一个多功能的数据结构服务器。它支持多种数据结构，如字符串（Strings）、哈希（Hashes）、列表（Lists）、集合（Sets）、有序集合（Sorted Sets）等，并提供了丰富的原子操作，使其在缓存、会话管理、消息队列、实时排行榜、计数器等场景中表现出色。

Python作为一门简洁、高效、功能强大的编程语言，在Web开发、数据科学、自动化运维等领域拥有广泛的应用。当Python与Redis相结合时，便能擦出无限的火花，帮助开发者构建出高性能、可伸缩的应用程序。本文将从零开始，详细介绍如何使用Python操作Redis，从环境搭建到核心数据结构的操作，再到高级特性的应用，力求为读者提供一份全面、深入的指南。

第一章：环境准备与基本连接

在开始Python操作Redis之前，我们首先需要准备好运行环境。

1.1 安装Redis服务器

Redis服务器可以在多种操作系统上运行。

Linux/macOS:
推荐使用包管理器安装：
“`bash
# Debian/Ubuntu
sudo apt update
sudo apt install redis-server

CentOS/RHEL

sudo yum install epel-release
sudo yum install redis

macOS (使用Homebrew)

brew install redis
安装完成后，可以通过以下命令启动Redis服务：bash
redis-server
`` 如果想在后台运行，可以修改配置文件/etc/redis/redis.conf(Linux) 或/usr/local/etc/redis.conf(macOS)，将daemonize no改为daemonize yes`。
Windows:
Redis官方不直接支持Windows，但Microsoft Open Technologies团队提供了一个Win64版本。你可以从GitHub releases页面下载：https://github.com/microsoftarchive/redis/releases。下载后解压，运行 redis-server.exe 即可。

无论何种方式，启动Redis服务器后，可以通过 redis-cli 工具进行测试：
bash redis-cli ping
如果返回 PONG，则表示Redis服务器已成功运行。

1.2 安装Python Redis客户端库

Python操作Redis需要一个官方推荐的客户端库——redis-py。使用pip进行安装非常简单：
bash pip install redis

1.3 建立与Redis的连接

安装完redis-py后，我们就可以在Python代码中与Redis建立连接了。

“`python
import redis

默认连接到 localhost:6379, 数据库0

host参数指定Redis服务器地址，port指定端口，db指定数据库索引（默认为0-15）

r = redis.Redis(host=’localhost’, port=6379, db=0)

try:
# 使用ping()方法测试连接是否成功
if r.ping():
print(“成功连接到Redis服务器！”)
else:
print(“无法连接到Redis服务器。”)
except redis.exceptions.ConnectionError as e:
print(f”连接错误: {e}”)

当然，也可以通过URL方式连接，方便管理

r = redis.from_url(‘redis://localhost:6379/0’)

“`

连接池（Connection Pool）:
在高并发或长期运行的应用中，为每个请求创建新的连接会耗费资源。redis-py 提供了连接池功能，可以复用连接，提高性能。

“`python
import redis

创建一个连接池，指定最大连接数等参数

pool = redis.ConnectionPool(host=’localhost’, port=6379, db=0, max_connections=10)

从连接池获取连接

r = redis.Redis(connection_pool=pool)

try:
if r.ping():
print(“成功通过连接池连接到Redis服务器！”)
except redis.exceptions.ConnectionError as e:
print(f”连接错误: {e}”)

应用程序结束时，可以关闭连接池（可选，Python的垃圾回收机制通常会处理）

pool.disconnect()

“`

推荐在生产环境中使用连接池，因为它能有效管理连接资源，避免频繁地创建和销毁TCP连接。

第二章：Redis核心数据结构操作

Redis支持五种基本数据结构：字符串、哈希、列表、集合和有序集合。理解并熟练操作它们是掌握Redis的关键。

2.1 字符串 (Strings)

字符串是Redis最基本的数据类型，可以存储文本、数字、二进制数据等。最大容量为512MB。

“`python

设置键值对

r.set(‘name’, ‘Alice’)
r.set(‘age’, 30)
r.set(‘city’, ‘New York’)

获取键的值

name = r.get(‘name’)
print(f”Name: {name.decode(‘utf-8’)}”) # Redis返回的是字节串，需要解码

批量设置和获取

r.mset({‘country’: ‘USA’, ‘continent’: ‘North America’})
data = r.mget([‘name’, ‘city’, ‘country’])
print(f”Batch data: {[d.decode(‘utf-8’) if d else None for d in data]}”)

数值操作：递增/递减

r.set(‘views’, 100)
r.incr(‘views’) # 递增1
r.incrby(‘views’, 5) # 递增5
r.decr(‘views’) # 递减1
r.decrby(‘views’, 3) # 递减3
print(f”Views: {r.get(‘views’).decode(‘utf-8’)}”)

设置带过期时间的键 (seconds)

r.setex(‘temp_key’, 10, ‘This will expire in 10 seconds’) # 10秒后过期
print(f”Temp Key: {r.get(‘temp_key’)}”)
“`

2.2 哈希 (Hashes)

哈希类似于Python字典，可以存储多个字段-值对，适用于存储对象（如用户信息）。

“`python

设置哈希字段

r.hset(‘user:1’, ‘name’, ‘Bob’)
r.hset(‘user:1′, ’email’, ‘[email protected]’)
r.hset(‘user:1’, ‘age’, 25)

批量设置哈希字段

r.hmset(‘user:2’, {‘name’: ‘Charlie’, ’email’: ‘[email protected]’, ‘age’: 35})

获取单个字段

user_name = r.hget(‘user:1’, ‘name’)
print(f”User 1 Name: {user_name.decode(‘utf-8’)}”)

获取所有字段和值

user_data = r.hgetall(‘user:1’)
print(“User 1 Data:”)
for field, value in user_data.items():
print(f” {field.decode(‘utf-8’)}: {value.decode(‘utf-8’)}”)

获取多个指定字段

user2_info = r.hmget(‘user:2’, ‘name’, ‘age’)
print(f”User 2 Info: {[i.decode(‘utf-8’) for i in user2_info]}”)

删除字段

r.hdel(‘user:1’, ‘age’)
print(f”User 1 Age after deletion: {r.hget(‘user:1’, ‘age’)}”) # None
“`

2.3 列表 (Lists)

列表是元素的有序集合，可以从头部或尾部添加/移除元素，适用于实现队列（FIFO）或栈（LIFO）。

“`python

左侧插入 (栈的push，队列的enqueue)

r.lpush(‘my_list’, ‘element A’)
r.lpush(‘my_list’, ‘element B’)
r.rpush(‘my_list’, ‘element C’) # 右侧插入

查看列表所有元素 (从头到尾，索引0到-1表示所有)

list_elements = r.lrange(‘my_list’, 0, -1)
print(f”List elements: {[e.decode(‘utf-8’) for e in list_elements]}”)

期望输出: [‘element B’, ‘element A’, ‘element C’]

左侧弹出 (栈的pop，队列的dequeue)

popped_left = r.lpop(‘my_list’)
print(f”Popped from left: {popped_left.decode(‘utf-8’)}”)

期望输出: element B

右侧弹出

popped_right = r.rpop(‘my_list’)
print(f”Popped from right: {popped_right.decode(‘utf-8’)}”)

期望输出: element C

获取列表长度

list_length = r.llen(‘my_list’)
print(f”List length: {list_length}”)

期望输出: 1 (只剩下element A)

“`

2.4 集合 (Sets)

集合是无序的、不重复元素的集合，适用于存储唯一项或进行集合运算（交集、并集、差集）。

“`python

添加元素到集合

r.sadd(‘tags’, ‘python’, ‘redis’, ‘database’)
r.sadd(‘tags’, ‘python’) # 重复添加不会生效

获取集合所有成员

all_tags = r.smembers(‘tags’)
print(f”All tags: {[t.decode(‘utf-8’) for t in all_tags]}”)

注意：输出顺序不确定

检查元素是否在集合中

is_member = r.sismember(‘tags’, ‘redis’)
print(f”Is ‘redis’ in tags? {is_member}”)

移除元素

r.srem(‘tags’, ‘database’)
print(f”Tags after removal: {[t.decode(‘utf-8’) for t in r.smembers(‘tags’)]}”)

集合运算

r.sadd(‘developers’, ‘Alice’, ‘Bob’, ‘Charlie’, ‘python’)
r.sadd(‘designers’, ‘Charlie’, ‘David’, ‘Eve’, ‘database’)

交集 (共同的成员)

common_members = r.sinter(‘tags’, ‘developers’)
print(f”Common tags and developers: {[m.decode(‘utf-8’) for m in common_members]}”)

并集 (所有不重复的成员)

all_members = r.sunion(‘developers’, ‘designers’)
print(f”All developers and designers: {[m.decode(‘utf-8’) for m in all_members]}”)

差集 (在第一个集合中但不在第二个集合中的成员)

dev_only = r.sdiff(‘developers’, ‘designers’)
print(f”Developers only: {[m.decode(‘utf-8’) for m in dev_only]}”)
“`

2.5 有序集合 (Sorted Sets)

有序集合是集合的扩展，每个成员都关联一个分数（score），集合中的成员是唯一的，但分数可以重复。元素根据分数进行排序。适用于排行榜、带权重的标签等。

“`python

添加成员到有序集合 (score, member)

r.zadd(‘leaderboard’, {‘Alice’: 100, ‘Bob’: 85, ‘Charlie’: 120})
r.zadd(‘leaderboard’, {‘David’: 95}) # 添加新成员

获取指定范围的成员 (按分数从小到大)

withscores=True 可以同时获取分数

top_players = r.zrange(‘leaderboard’, 0, -1, withscores=True)
print(“Top players (ascending):”)
for member, score in top_players:
print(f” {member.decode(‘utf-8’)}: {score}”)

获取指定范围的成员 (按分数从大到小)

top_players_desc = r.zrevrange(‘leaderboard’, 0, -1, withscores=True)
print(“Top players (descending):”)
for member, score in top_players_desc:
print(f” {member.decode(‘utf-8’)}: {score}”)

获取成员的分数

score_bob = r.zscore(‘leaderboard’, ‘Bob’)
print(f”Bob’s score: {score_bob}”)

增加成员的分数

r.zincrby(‘leaderboard’, 15, ‘Bob’)
print(f”Bob’s new score: {r.zscore(‘leaderboard’, ‘Bob’)}”)

获取指定分数范围的成员

high_scores = r.zrangebyscore(‘leaderboard’, 90, 110, withscores=True)
print(f”Players with scores between 90 and 110: {[ (m.decode(‘utf-8’), s) for m, s in high_scores]}”)

移除成员

r.zrem(‘leaderboard’, ‘Bob’)
print(f”Leaderboard after Bob’s removal: {[ (m.decode(‘utf-8’), s) for m,s in r.zrange(‘leaderboard’, 0, -1, withscores=True)]}”)
“`

第三章：高级特性与实战技巧

除了基本数据结构操作，Redis还提供了许多高级特性，可以帮助我们构建更健壮、高效的应用。

3.1 键管理与过期时间

Redis中的所有数据都存储在键值对中，合理管理键的生命周期至关重要。

“`python

检查键是否存在

exists = r.exists(‘name’)
print(f”Does ‘name’ exist? {exists}”)

获取所有键 (生产环境不推荐，会阻塞服务器)

keys = r.keys(‘*’)

print(f”All keys: {[k.decode(‘utf-8’) for k in keys]}”)

删除键

r.delete(‘name’, ‘age’, ‘country’)
print(f”Does ‘name’ exist after deletion? {r.exists(‘name’)}”)

设置键的过期时间 (秒)

r.set(‘cache_data’, ‘some_value’)
r.expire(‘cache_data’, 60) # 60秒后过期

设置带过期时间的键 (更推荐的写法)

r.setex(‘temp_session_id’, 3600, ‘user_token_abc’) # 1小时过期

获取键的剩余生存时间 (TTL – Time To Live)

-1 表示永不过期，-2 表示键不存在

ttl = r.ttl(‘temp_session_id’)
print(f”TTL of ‘temp_session_id’: {ttl} seconds”)

移除键的过期时间，使其永不过期

r.persist(‘temp_session_id’)
print(f”New TTL of ‘temp_session_id’: {r.ttl(‘temp_session_id’)}”)
“`

3.2 事务 (Transactions) 与管道 (Pipelining)

Redis是单线程的，所有命令都是原子性的。但如果想执行一组命令，确保它们作为一个原子操作执行（要么全部成功，要么全部失败），或者批量发送命令以提高效率，就需要用到事务和管道。

管道 (Pipelining):
redis-py 的管道功能允许客户端一次性向Redis服务器发送多个命令，然后一次性读取所有回复，从而减少网络往返时间（RTT），提高性能。它不是事务性的，即中间的命令失败不会影响后续命令的执行。

python pipe = r.pipeline() # 创建管道 pipe.set('key1', 'value1') pipe.incr('counter') pipe.sadd('myset', 'a', 'b', 'c') results = pipe.execute() # 执行所有命令并获取结果 print(f"Pipeline results: {results}")
事务 (MULTI/EXEC):
Redis事务通过 MULTI 和 EXEC 命令实现。MULTI 开启一个事务块，之后的所有命令都会被放入队列，直到 EXEC 命令被调用，所有命令才会原子性地执行。如果在 EXEC 之前有任何命令失败，整个事务通常会回滚（但Redis的事务回滚机制与传统数据库不同，它只保证命令的原子执行，不保证失败回滚）。

redis-py 的管道默认就支持事务：当你调用 pipeline() 时，它会自动启用事务行为 (MULTI/EXEC)。如果你不希望是事务性的，可以传入 transaction=False。

“`python

默认是事务性的

pipe_tx = r.pipeline()
pipe_tx.set(‘balance’, 100)
pipe_tx.incrby(‘balance’, 10)
pipe_tx.decrby(‘balance’, 5)
tx_results = pipe_tx.execute()
print(f”Transaction results: {tx_results}”)
print(f”Final balance: {r.get(‘balance’).decode(‘utf-8’)}”)

WATCH 命令用于乐观锁，在MULTI/EXEC执行期间监控键的变化

如果在WATCH的键被修改，EXEC会返回None，表示事务失败

with r.pipeline() as pipe:
while True:
try:
pipe.watch(‘account_balance’) # 监控account_balance键
current_balance = int(pipe.get(‘account_balance’) or 0)
if current_balance < 50:
print(“余额不足，事务取消。”)
pipe.unwatch() # 取消监控
break
```
        pipe.multi() # 标记事务开始
        pipe.set('account_balance', current_balance - 50)
        pipe.set('last_transaction', 'withdrawal 50')
        pipe.execute() # 执行事务
        print("取款成功！")
        break
    except redis.exceptions.WatchError:
        print("余额已被修改，重试事务。")
        continue # 重试
```
“`

3.3 发布/订阅 (Pub/Sub)

Redis的发布/订阅系统允许客户端订阅一个或多个频道，当有消息发布到这些频道时，订阅者会立即收到消息。这非常适合实现实时聊天、通知系统、事件驱动架构等。

“`python

订阅者代码 (通常在一个独立的进程或线程中运行)

def subscriber():
pubsub = r.pubsub()
pubsub.subscribe(‘chat_channel’, ‘news_updates’) # 订阅多个频道
print(“订阅者已启动，等待消息…”)
for message in pubsub.listen():
if message[‘type’] == ‘message’:
channel = message[‘channel’].decode(‘utf-8’)
data = message[‘data’].decode(‘utf-8’)
print(f”收到来自频道 ‘{channel}’ 的消息: {data}”)
elif message[‘type’] == ‘subscribe’:
print(f”成功订阅频道: {message[‘channel’].decode(‘utf-8’)}”)

发布者代码

def publisher():
import time
print(“发布者启动…”)
r.publish(‘chat_channel’, ‘Hello everyone!’)
time.sleep(1)
r.publish(‘news_updates’, ‘Breaking news: Redis is awesome!’)
time.sleep(1)
r.publish(‘chat_channel’, ‘How are you doing?’)
time.sleep(1)
r.publish(‘news_updates’, ‘Redis 7.0 released!’)
print(“发布者发送完毕。”)

为了演示，我们可以在主线程中启动一个子线程作为订阅者，然后执行发布者

import threading
sub_thread = threading.Thread(target=subscriber)
sub_thread.daemon = True # 设置为守护线程，主程序退出时自动结束
sub_thread.start()

等待订阅者启动并订阅频道

import time
time.sleep(2) # 给订阅者一些时间启动

publisher()
“`

3.4 持久化 (Persistence)

尽管Redis是内存数据库，但它提供了两种持久化方式来防止数据丢失：

RDB (Redis Database): 定期将内存中的数据快照写入磁盘。
AOF (Append Only File): 将所有写命令追加到文件中。

这些配置通常在 redis.conf 文件中设置，redis-py 客户端不需要直接操作持久化，但理解其工作原理对部署和数据恢复至关重要。

第四章：Redis实战应用场景

Redis的强大之处在于其多功能性，使其能胜任多种应用场景。

4.1 作为缓存层

这是Redis最常见的用途。将频繁访问的数据存储在Redis中，减少对数据库的访问压力。

“`python
import json

def get_user_data(user_id):
# 尝试从Redis缓存中获取
cached_data = r.get(f’user:{user_id}:cache’)
if cached_data:
print(f”从缓存获取用户 {user_id} 数据。”)
return json.loads(cached_data.decode(‘utf-8’))

# 如果缓存中没有，则从数据库查询 (模拟)
print(f"从数据库获取用户 {user_id} 数据。")
# 模拟数据库查询延迟
import time
time.sleep(0.5)
user_data = {'id': user_id, 'name': f'User {user_id}', 'email': f'user{user_id}@example.com'}

# 将数据存入Redis缓存，设置过期时间（例如5分钟）
r.setex(f'user:{user_id}:cache', 300, json.dumps(user_data))
return user_data

第一次查询，从数据库获取并写入缓存

print(get_user_data(101))

第二次查询，从缓存获取

print(get_user_data(101))
“`

4.2 会话管理

将用户会话信息存储在Redis中，实现无状态的Web应用，方便水平扩展。

“`python
import uuid

def create_session(user_id):
session_id = str(uuid.uuid4())
session_data = {‘user_id’: user_id, ‘login_time’: time.time()}
# 使用哈希存储会话数据，设置过期时间为30分钟
r.hmset(f’session:{session_id}’, session_data)
r.expire(f’session:{session_id}’, 1800)
print(f”创建会话: {session_id} for user {user_id}”)
return session_id

def get_session(session_id):
session_data = r.hgetall(f’session:{session_id}’)
if session_data:
# 刷新会话过期时间
r.expire(f’session:{session_id}’, 1800)
decoded_data = {k.decode(‘utf-8’): v.decode(‘utf-8’) for k, v in session_data.items()}
print(f”获取会话: {session_id} – {decoded_data}”)
return decoded_data
print(f”会话 {session_id} 不存在或已过期。”)
return None

session_id = create_session(201)
time.sleep(1)
get_session(session_id)

模拟等待会话过期

time.sleep(1801)

get_session(session_id)

“`

4.3 消息队列

Redis列表可以作为简单的消息队列，实现生产者-消费者模式。

“`python

生产者

def producer(message):
r.rpush(‘message_queue’, message) # 消息从右侧入队
print(f”发送消息: {message}”)

消费者

def consumer():
print(“消费者启动，等待消息…”)
while True:
# blpop阻塞式从左侧出队，等待1秒，如果没有消息则返回None
# ‘message_queue’ 是队列名，1是超时时间
item = r.blpop(‘message_queue’, timeout=1)
if item:
queue_name, message = item
print(f”收到消息: {message.decode(‘utf-8’)}”)
else:
print(“没有新消息，等待中…”)
# break # 实际应用中可能不会break，而是持续监听

启动消费者线程

consumer_thread = threading.Thread(target=consumer)
consumer_thread.daemon = True
consumer_thread.start()

生产者发送消息

producer(“任务1：处理图片”)
time.sleep(0.5)
producer(“任务2：发送邮件”)
time.sleep(0.5)
producer(“任务3：更新数据”)
time.sleep(3) # 留足够时间让消费者处理
“`

4.4 排行榜/计数器

有序集合天然适合做排行榜，字符串的 INCR 系列命令适合做计数器。

“`python

计数器

r.set(‘page_views’, 0)
r.incr(‘page_views’)
r.incrby(‘page_views’, 10)
print(f”Current page views: {r.get(‘page_views’).decode(‘utf-8’)}”)

排行榜

r.zadd(‘game_scores’, {‘PlayerA’: 1500, ‘PlayerB’: 1200, ‘PlayerC’: 1800, ‘PlayerD’: 1350})
r.zincrby(‘game_scores’, 200, ‘PlayerA’) # PlayerA得分增加200

print(“\n当前排行榜 (前3名):”)
top_3 = r.zrevrange(‘game_scores’, 0, 2, withscores=True)
for rank, (player, score) in enumerate(top_3):
print(f” No.{rank+1} – {player.decode(‘utf-8’)}: {score}”)
“`

第五章：最佳实践与注意事项

5.1 错误处理

在实际应用中，网络连接中断、Redis服务器宕机等情况时有发生，务必进行错误处理。

python try: r.set('test_key', 'test_value') result = r.get('test_key') print(f"Operation successful: {result.decode('utf-8')}") except redis.exceptions.ConnectionError as e: print(f"Redis连接错误: {e}") except redis.exceptions.RedisError as e: print(f"Redis操作错误: {e}") except Exception as e: print(f"发生未知错误: {e}")

5.2 内存管理

Redis是内存数据库，内存是其核心资源。

INFO memory: 使用 redis-cli INFO memory 命令监控内存使用情况。
MAXMEMORY: 在 redis.conf 中配置 maxmemory 参数，限制Redis可使用的最大内存。
MAXMEMORY-POLICY: 配置内存淘汰策略（如LRU、LFU等），当内存达到上限时，Redis如何选择键进行删除。allkeys-lru 是常用且推荐的策略。
避免大键: 尽量避免存储过大的字符串、列表、哈希等，这可能导致网络延迟和内存碎片。
合理设置过期时间: 对缓存数据设置合理的过期时间，及时释放内存。

5.3 安全性

密码保护: 在 redis.conf 中设置 requirepass your_password，并在Python连接时提供密码：r = redis.Redis(host='localhost', port=6379, password='your_password')。
绑定IP: 使用 bind 127.0.0.1 限制Redis只能被本地访问，或者只绑定内网IP。
禁用危险命令: 通过 rename-command 或 config set 禁用或重命名像 KEYS、FLUSHALL 这样的命令。

5.4 性能优化

使用连接池: 前面已提到，这是最基本的优化。
管道/事务: 批量操作，减少网络往返。
选择正确的数据结构: 根据业务场景选择最适合的Redis数据结构。例如，存储对象用哈希比多个字符串更高效。
数据序列化: 对于非字符串数据，选择高效的序列化方式（如JSON、MsgPack、Protocol Buffers），并在存取时进行编解码。
避免 KEYS 命令: KEYS 会遍历所有键，在大规模数据集上可能导致Redis阻塞。应使用 SCAN 命令进行迭代式查找。

5.5 监控与维护

INFO 命令: 提供Redis服务器的各种统计信息，包括内存、CPU、客户端连接数、命中率等。
Redis Sentinel / Redis Cluster: 对于高可用和可伸缩性需求，考虑使用Redis Sentinel（高可用）或Redis Cluster（分布式）。redis-py 也支持连接这些部署模式。

结语

至此，我们已经详细探讨了从零开始使用Python操作Redis的方方面面，从环境搭建、基础数据结构操作，到高级特性如事务、发布/订阅，再到实战应用场景和最佳实践。

Redis凭借其卓越的性能和丰富的数据结构，已成为现代应用程序不可或缺的一部分。掌握Python与Redis的结合，将极大地提升你在构建高性能、可伸缩系统时的能力。希望这份指南能为你打开Redis世界的大门，让你在未来的开发实践中游刃有余。

记住，实践是最好的老师。现在就动手尝试，将所学知识应用到你的项目中，不断探索Redis的更多可能性吧！

从零开始：Python操作Redis完全指南

引言：Redis的魅力与Python的结合

第一章：环境准备与基本连接

1.1 安装Redis服务器

CentOS/RHEL

macOS (使用Homebrew)

1.2 安装Python Redis客户端库

1.3 建立与Redis的连接

默认连接到 localhost:6379, 数据库0

host参数指定Redis服务器地址，port指定端口，db指定数据库索引（默认为0-15）

当然，也可以通过URL方式连接，方便管理

r = redis.from_url(‘redis://localhost:6379/0’)

创建一个连接池，指定最大连接数等参数

从连接池获取连接

应用程序结束时，可以关闭连接池（可选，Python的垃圾回收机制通常会处理）

pool.disconnect()

第二章：Redis核心数据结构操作

2.1 字符串 (Strings)

设置键值对

获取键的值

批量设置和获取

数值操作：递增/递减

设置带过期时间的键 (seconds)

2.2 哈希 (Hashes)

设置哈希字段

批量设置哈希字段

获取单个字段

获取所有字段和值

获取多个指定字段

删除字段

2.3 列表 (Lists)

左侧插入 (栈的push，队列的enqueue)

查看列表所有元素 (从头到尾，索引0到-1表示所有)

期望输出: [‘element B’, ‘element A’, ‘element C’]

左侧弹出 (栈的pop，队列的dequeue)

期望输出: element B

右侧弹出

期望输出: element C

获取列表长度

期望输出: 1 (只剩下element A)

2.4 集合 (Sets)

添加元素到集合

获取集合所有成员

注意：输出顺序不确定

检查元素是否在集合中

移除元素

集合运算

交集 (共同的成员)

并集 (所有不重复的成员)

差集 (在第一个集合中但不在第二个集合中的成员)

2.5 有序集合 (Sorted Sets)

添加成员到有序集合 (score, member)

获取指定范围的成员 (按分数从小到大)

withscores=True 可以同时获取分数

获取指定范围的成员 (按分数从大到小)

获取成员的分数

增加成员的分数

获取指定分数范围的成员

移除成员

第三章：高级特性与实战技巧

3.1 键管理与过期时间

检查键是否存在

获取所有键 (生产环境不推荐，会阻塞服务器)

keys = r.keys(‘*’)

print(f”All keys: {[k.decode(‘utf-8’) for k in keys]}”)

删除键

设置键的过期时间 (秒)

设置带过期时间的键 (更推荐的写法)

获取键的剩余生存时间 (TTL – Time To Live)

-1 表示永不过期，-2 表示键不存在

移除键的过期时间，使其永不过期

3.2 事务 (Transactions) 与管道 (Pipelining)

默认是事务性的

WATCH 命令用于乐观锁，在MULTI/EXEC执行期间监控键的变化

如果在WATCH的键被修改，EXEC会返回None，表示事务失败

3.3 发布/订阅 (Pub/Sub)

订阅者代码 (通常在一个独立的进程或线程中运行)

发布者代码

为了演示，我们可以在主线程中启动一个子线程作为订阅者，然后执行发布者

等待订阅者启动并订阅频道

发表评论取消回复