当前位置：首页 > 文章列表 > 文章 > python教程 > Python操作Redis和Memcached教程

Python操作Redis和Memcached教程

2025-09-13 08:52:32 0浏览收藏

本篇文章给大家分享《Python操作Redis与Memcached教程》，覆盖了文章的常见基础知识，其实一个语言的全部知识点一篇文章是不可能说完的，但希望通过这些问题，让读者对自己的掌握程度有一定的认识(B 数)，从而弥补自己的不足，更好的掌握它。

答案：Python操作Redis和Memcached需使用redis-py和python-memcached库，通过连接池、管道、序列化优化性能，Redis适合复杂数据结构与持久化场景，Memcached适用于高性能键值缓存，高可用需结合哨兵、集群或客户端分片。

如何使用Python操作Redis/Memcached？

在Python中操作Redis和Memcached，主要依赖各自成熟的客户端库。对于Redis，我们通常会使用redis-py这个库，它提供了非常全面且直观的API来与Redis服务器交互，支持各种数据结构和高级特性。而对于Memcached，python-memcached则是事实上的标准，虽然Memcached的功能相对简单，但这个库也足以满足日常的缓存需求。它们的核心思路都是通过网络协议与服务器建立连接，然后将Python对象序列化为字节流发送给服务器，再将服务器返回的字节流反序列化回Python对象。

解决方案

要使用Python操作Redis或Memcached，你需要先安装对应的客户端库。

操作Redis：

安装redis-py：
```
pip install redis
```

基本使用示例：redis-py提供了Redis和StrictRedis两个类。StrictRedis更推荐，因为它遵循Redis的命令规范，不会做任何隐式的类型转换。

import redis

# 连接到Redis服务器
# 默认连接到localhost:6379，db=0
# 如果Redis有密码，可以这样连接：
# r = redis.StrictRedis(host='localhost', port=6379, db=0, password='your_password')
try:
    r = redis.StrictRedis(host='localhost', port=6379, db=0, decode_responses=True)
    # decode_responses=True 会自动将Redis返回的字节流解码为UTF-8字符串，省去手动解码的麻烦

    # 字符串操作
    r.set('mykey', 'Hello Redis!')
    value = r.get('mykey')
    print(f"Key 'mykey': {value}") # 输出: Key 'mykey': Hello Redis!

    r.setex('temp_key', 10, 'This will expire in 10 seconds') # 设置带过期时间的键

    # 哈希操作
    r.hset('user:100', mapping={'name': 'Alice', 'age': 30, 'city': 'New York'})
    user_info = r.hgetall('user:100')
    print(f"User 100 info: {user_info}") # 输出: User 100 info: {'name': 'Alice', 'age': '30', 'city': 'New York'}

    # 列表操作
    r.rpush('mylist', 'item1', 'item2', 'item3')
    list_items = r.lrange('mylist', 0, -1)
    print(f"List 'mylist': {list_items}") # 输出: List 'mylist': ['item1', 'item2', 'item3']

    # 集合操作
    r.sadd('myset', 'apple', 'banana', 'apple') # 'apple'只会被添加一次
    set_members = r.smembers('myset')
    print(f"Set 'myset': {set_members}") # 输出: Set 'myset': {'apple', 'banana'}

    # 有序集合操作
    r.zadd('myzset', {'member1': 10, 'member2': 5, 'member3': 15})
    sorted_members = r.zrange('myzset', 0, -1, withscores=True)
    print(f"Sorted Set 'myzset': {sorted_members}") # 输出: Sorted Set 'myzset': [('member2', 5.0), ('member1', 10.0), ('member3', 15.0)]

    # 删除键
    r.delete('mykey', 'user:100', 'mylist', 'myset', 'myzset', 'temp_key')
    print("Keys deleted.")

except redis.exceptions.ConnectionError as e:
    print(f"无法连接到Redis服务器: {e}")
except Exception as e:
    print(f"发生错误: {e}")

操作Memcached：

安装python-memcached：
```
pip install python-memcached
```

基本使用示例： Memcached的API相对简单，主要就是set、get和delete。

import memcache

# 连接到Memcached服务器
# 可以传入一个服务器地址列表，实现客户端分片
mc = memcache.Client(['127.0.0.1:11211'], debug=0) # debug=0 关闭调试信息

try:
    # 设置键值对
    # set(key, value, time=0)
    # time参数是过期时间，单位秒。0表示永不过期。
    mc.set('my_mem_key', 'Hello Memcached!', time=60) # 键会在60秒后过期
    print("Set 'my_mem_key'.")

    # 获取键值对
    value = mc.get('my_mem_key')
    if value:
        print(f"Key 'my_mem_key': {value}") # 输出: Key 'my_mem_key': Hello Memcached!
    else:
        print("Key 'my_mem_key' not found or expired.")

    # 添加键值对（如果键不存在则添加）
    mc.add('new_key', 'This is a new value', time=30)
    print(f"Added 'new_key': {mc.get('new_key')}")

    # 替换键值对（如果键存在则替换）
    mc.replace('new_key', 'This is a replaced value', time=30)
    print(f"Replaced 'new_key': {mc.get('new_key')}")

    # 删除键
    mc.delete('my_mem_key')
    print("Deleted 'my_mem_key'.")
    print(f"After deletion, 'my_mem_key': {mc.get('my_mem_key')}") # 输出: After deletion, 'my_mem_key': None

except Exception as e:
    print(f"操作Memcached时发生错误: {e}")

Python操作Redis时有哪些常见的陷阱或性能优化策略？

在使用Python与Redis交互时，我个人觉得，虽然redis-py设计得很棒，但一些细节处理不好确实会掉进坑里，或者说，会白白浪费掉Redis本该有的高性能。我总结了一些经验，觉得这几点是开发者需要特别留意的。

首先，连接管理是重中之重。你可能觉得每次操作都创建一个新的Redis连接没什么大不了，但实际上，建立和关闭TCP连接的开销是显著的。在并发量大的应用中，这会导致大量的TIME_WAIT状态，甚至耗尽文件描述符，最终引发Too many open files的错误。所以，使用连接池（Connection Pool）几乎是强制性的。redis-py提供了redis.ConnectionPool，你只需在初始化客户端时传入它，库就会自动管理连接的复用，大大降低了连接开销。

import redis

# 创建一个连接池
pool = redis.ConnectionPool(host='localhost', port=6379, db=0, decode_responses=True)
r = redis.StrictRedis(connection_pool=pool)

# 之后所有的r操作都会从连接池中获取和释放连接
r.set('foo', 'bar')
print(r.get('foo'))

其次，减少网络往返时间（RTT）。Redis是单线程的，但它通过多路复用I/O模型实现高并发。然而，如果你的Python应用频繁地向Redis发送单个命令，即使Redis处理速度再快，网络延迟也会成为瓶颈。这时候，管道（Pipeline）就显得尤为重要了。管道允许你一次性发送多个命令给Redis，Redis会按顺序执行它们，然后一次性返回所有结果。这就像你一次性把购物清单交给店员，而不是每拿一件商品就跑去结账一次。

# 使用管道
pipe = r.pipeline()
pipe.set('key1', 'value1')
pipe.set('key2', 'value2')
pipe.get('key1')
pipe.get('key2')
results = pipe.execute()
print(f"Pipeline results: {results}") # 输出: Pipeline results: [True, True, 'value1', 'value2']

对于需要原子性执行一系列命令的场景，管道结合MULTI和EXEC可以实现事务（Transaction）。这保证了在事务中的所有命令要么全部执行成功，要么全部不执行，并且在执行期间不会被其他客户端的命令插入。

再者，数据序列化。Redis存储的是字节串，当你需要存储Python对象（如字典、列表或自定义对象）时，就需要进行序列化。最常见的方式是使用JSON或pickle。JSON可读性好，跨语言兼容性强，但对复杂Python对象（如日期时间、自定义类实例）支持不佳。pickle则能序列化几乎所有Python对象，但它是Python特有的，且存在安全风险（反序列化恶意数据可能导致代码执行）。我的建议是，如果数据结构简单且需要跨语言，用JSON；如果仅限Python内部使用且对性能要求高，可以考虑pickle，但务必确保数据的来源是可信的。

import json
import pickle

data = {'name': 'Charlie', 'age': 25, 'interests': ['coding', 'reading']}

# JSON序列化
r.set('user:json', json.dumps(data))
retrieved_json = json.loads(r.get('user:json'))
print(f"JSON data: {retrieved_json}")

# Pickle序列化 (注意decode_responses=True可能会影响pickle的直接使用，因为它会先尝试解码)
# 如果使用pickle，通常会关闭decode_responses，手动处理字节
r_bytes = redis.StrictRedis(connection_pool=pool, decode_responses=False)
r_bytes.set('user:pickle', pickle.dumps(data))
retrieved_pickle = pickle.loads(r_bytes.get('user:pickle'))
print(f"Pickle data: {retrieved_pickle}")

最后，键名设计和过期策略也值得一提。清晰的键名设计（比如user:{id}:profile）不仅有助于管理，还能避免键冲突。而合理设置过期时间（TTL）是管理Redis内存、避免内存溢出的关键。对于那些非永久性的缓存数据，一定要设置合理的过期时间，否则你的Redis实例可能会无休止地增长下去。

Memcached与Redis在Python应用中，各自适合哪些场景？

选择Memcached还是Redis，这在我看来，就像选择一把锋利的菜刀还是一把瑞士军刀。两者都能切东西，但功能和适用场景却大相径庭。在Python应用中，我通常会根据项目的具体需求来做判断。

Memcached的适用场景：

Memcached就像那把锋利的菜刀，它的设计哲学就是纯粹、简单、极致的键值缓存。它不提供复杂的数据结构，不支持数据持久化，也没有Pub/Sub、事务这些高级功能。它的优势在于：

极高的读写性能： 对于简单的键值对存储和检索，Memcached的性能表现非常出色。它没有Redis那些复杂功能的开销，内存管理也相对直接。
分布式和横向扩展能力： Memcached天生就是为分布式缓存设计的。客户端库（如python-memcached）通常会内置哈希算法，将不同的键分散到不同的Memcached服务器上。这使得扩展非常方便，只需增加服务器即可。
简单会话存储： 比如Web应用的Session数据，如果可以接受会话丢失（例如服务器重启），Memcached是一个很好的选择。
页面/API响应缓存： 对于那些计算成本高、但内容相对不常变化的页面或API响应，Memcached可以极大地提升响应速度。

我个人觉得，如果你的需求仅仅是“快速地存取一些简单的、可以丢失的键值对”，并且对内存占用有较高要求，那么Memcached无疑是更轻量、更直接的选择。它就像一个高效的内存字典，用起来没什么负担。

Redis的适用场景：

Redis则更像那把瑞士军刀，它不仅能做缓存，还能做更多的事情。它的核心优势在于：

丰富的数据结构： 除了字符串，Redis还支持列表（List）、哈希（Hash）、集合（Set）、有序集合（Sorted Set）等多种数据结构。这意味着你可以用它来构建更复杂的数据模型，比如用户关注列表（Set）、排行榜（Sorted Set）、消息队列（List/Stream）等。
数据持久化： Redis支持RDB（快照）和AOF（日志）两种持久化方式，这意味着即使服务器重启，数据也不会丢失。这对于需要缓存但又不能完全容忍数据丢失的场景至关重要。
高级功能： Redis提供了Pub/Sub（发布/订阅）模式、事务、Lua脚本、地理空间索引、HyperLogLog等一系列高级功能。这些功能让Redis不仅仅是一个缓存，更是一个多功能的内存数据库。
更广泛的应用场景：
- 消息队列： 可以用List或Stream实现简单的消息队列。
- 实时排行榜/计数器： 利用Sorted Set和原子操作，轻松实现实时更新的排行榜。
- 分布式锁： 利用SETNX命令实现分布式锁，解决并发问题。
- 社交网络： 存储用户关系、关注列表、新鲜事等。
- 全页缓存（带持久化需求）： 如果缓存数据不能丢失，或者需要更复杂的缓存逻辑。

我通常会倾向于在项目中使用Redis，即使初期需求只是简单的键值缓存。因为随着项目的发展，很可能你会发现需要列表、集合或者持久化等功能，这时候如果一开始就用了Redis，后续的扩展会平滑得多。Memcached虽然快，但它的功能边界非常清晰，一旦超出这个边界，你就需要引入其他组件来弥补，这反而增加了系统的复杂性。当然，这也不是绝对的，对于流量极其庞大、对纯粹键值性能有极致追求的场景，Memcached仍然是不可替代的。

如何在Python应用中确保Redis/Memcached操作的稳定性和高可用性？

确保缓存服务的稳定性和高可用性，对于任何生产环境的Python应用来说都是至关重要的。毕竟，缓存一旦失效或服务中断，很可能导致数据库压力骤增，甚至拖垮整个系统。我在这方面有一些思考，觉得不光要关注缓存服务本身，还得从应用层面做好防护。

对于Redis：

Redis提供了多种机制来保证高可用性，而Python客户端库也很好地支持了这些机制。

哨兵模式（Sentinel）： 这是我个人觉得最常用、也是最成熟的Redis高可用方案之一。Sentinel集群负责监控主从Redis实例，当主节点失效时，它会自动进行故障转移，将一个从节点提升为新的主节点。在Python应用中，你需要使用redis.Sentinel来连接。它会从Sentinel集群中获取当前主节点的地址，并在主节点切换时自动更新连接。这大大简化了应用层的逻辑，不需要你手动去处理主从切换。

from redis.sentinel import Sentinel

sentinels = [('localhost', 26379), ('localhost', 26380), ('localhost', 26381)]
# 'mymaster'是你在Sentinel配置中定义的主节点名称
sentinel = Sentinel(sentinels, socket_timeout=0.1)

# 获取主节点连接
master = sentinel.master_for('mymaster', decode_responses=True)
master.set('mykey', 'Hello from Sentinel master!')
print(f"Master: {master.get('mykey')}")

# 获取从节点连接（用于读操作）
slave = sentinel.slave_for('mymaster', decode_responses=True)
print(f"Slave: {slave.get('mykey')}")

集群模式（Cluster）： 对于需要处理海量数据和更高并发的场景，Redis Cluster提供了数据分片（sharding）和高可用性。数据被分散存储在多个主节点上，每个主节点可以有自己的从节点。Python客户端（如redis-py-cluster，虽然redis-py也内置了对集群的支持）能够智能地将命令路由到正确的节点。使用集群模式时，你主要关心如何配置集群，客户端库会帮你处理底层的连接和路由。
主从复制（Replication）： 这是Redis高可用的基础。主节点负责写操作，从节点复制主节点的数据，可以用于读操作分流和数据冗余。虽然主从复制本身不提供自动故障转移，但它是Sentinel和Cluster模式的基石。在Python应用中，你可以直接连接到主节点进行写操作，连接到从节点进行读操作（但需要自行管理读写分离逻辑）。

对于Memcached：

Memcached本身没有内置的高可用性机制（如主从、集群）。它的高可用性主要依赖于客户端的分片和应用的容错能力。

客户端分片/一致性哈希： python-memcached客户端允许你配置多个Memcached服务器地址。当客户端存储或获取数据时，它会根据键的哈希值，将请求路由到特定的Memcached服务器。如果某个服务器宕机，客户端会自动跳过该服务器，尝试连接其他可用的服务器。一致性哈希是更高级的分片策略，它能减少在服务器增减时缓存的失效量。
```
import memcache

# 客户端会根据键的哈希值，将请求路由到这些服务器
# 如果其中一个服务器宕机，客户端会自动跳过
mc = memcache.Client(['10.0.0.1:11211', '10.0.0.2:11211', '10.0.0.3:11211'])
```
应用层容错： 由于Memcached是纯粹的缓存，它的数据被认为是“可丢失的”。这意味着，如果Memcached服务完全不可用，你的应用应该能够优雅地降级，例如直接从数据库获取数据，而不是直接报错。这种“缓存穿透”到数据库的行为，虽然会增加数据库压力，但至少保证了服务的可用性。

通用的稳定性和高可用性策略（适用于Redis和Memcached）：