Python使用Redis缓存

前言

最近编写了一个Flask应用，其API接口需要访问MySQL数据库读取数据，为了防止高并发情况下MySQL数据库的压力，使用了Redis作为缓存来提高API接口的响应速度。本文将介绍Redis的基本概念和在Python中的使用方法。

Redis简介

Redis（Remote Dictionary Server）是一个开源的高性能键值对存储数据库，最初由 Salvatore Sanfilippo 开发，它在内存中存储数据，并提供了持久化功能，可以将数据保存到磁盘中，是一种NoSQL(not-only sql,非关系型数据库)的数据库。

它的数据结构十分丰富，基础数据类型包括：string（字符串）、list（列表，双向链表）、hash（散列，键值对集合）、set（集合，不重复）和 sorted set（有序集合），这使得它不仅仅是一个简单的键值存储，还可以用于存储和处理复杂的数据。

Redis 特点/优势

Redis 具备许多特点和优势，所以在大规模应用和高并发场景中得到广泛应用。

  • 丰富的数据结构：Redis 支持多种数据结构，包括字符串（String）、哈希（Hash）、列表（List）、集合（Set）、有序集合（Sorted Set）等。每种数据结构都有丰富的操作命令，可以方便地对数据进行存储和处理。
  • 持久化：Redis 提供了两种持久化方式，RDB（Redis Database）和 AOF（Append Only File）。RDB 是将数据库在指定时间点转储到磁盘的快照持久化方式，AOF 是将写操作追加到文件末尾的日志持久化方式。这两种方式可以根据需求进行选择，保证数据在服务器重启后不会丢失。
  • 高性能：Redis 是一种内存数据库，数据存储在内存中，因此读写速度非常快。它采用单线程模型，避免了多线程带来的竞争问题，使得 Redis 能够充分利用 CPU 和内存资源。
  • 支持事务：Redis 支持事务操作，可以将多个命令打包执行，保证这些命令要么全部执行成功，要么全部失败，保持数据的一致性。虽然 Redis 在单个命令的执行上是原子性的，但是多个命令的组合并不是原子性的，通过事务可以实现一组命令的原子性执行。
  • 高可用与分布式：Redis 支持主从复制、哨兵和集群等功能，可以构建高可用和分布式的 Redis 架构。主从复制可以实现数据的热备份和读写分离，哨兵可以监控 Redis 的健康状态并进行自动故障转移，集群可以将数据分布在多个节点上，提高性能和扩展性。
  • 发布订阅：Redis 支持发布订阅模式，可以实现消息的发布和订阅。发布者将消息发布到指定的频道，订阅者可以订阅感兴趣的频道并接收相应的消息，实现了解耦和实时通信。

Redis 为什么很快

  • 内存存储：Redis将数据存储在内存中，而不是像传统的磁盘存储数据库那样将数据写入到硬盘上。由于内存的读写速度远远快于磁盘，因此Redis能够实现极快的读写性能。
  • 单线程模型：Redis采用单线程模型，每个Redis实例都由单个主线程来处理所有的客户端请求。虽然单线程看起来似乎会限制其性能，但这实际上是Redis的一大优势。单线程模型消除了多线程之间的竞争和锁等开销，使得Redis能够充分利用CPU资源，并且避免了多线程带来的复杂性。此外，Redis在内部使用了I/O多路复用技术（例如epoll或kqueue）来处理并发请求，使得单线程能够同时处理多个客户端连接。
  • 非阻塞IO：Redis使用了非阻塞IO，也就是在读写操作时不会阻塞其他操作。在读取数据时，如果内存中没有所需的数据，Redis会立即返回一个空结果，而不会等待数据从磁盘加载进来。这样即使在高并发情况下，Redis也能够快速地响应请求。
  • 高效的数据结构：Redis支持多种高效的数据结构，比如字符串、哈希、列表、集合、有序集合等。这些数据结构的设计和实现都非常高效，能够在常量时间内完成查找、插入、删除等操作，保证了Redis的高速性能。
  • 异步操作：Redis支持异步操作，比如异步持久化和异步复制。异步操作能够让Redis在进行磁盘持久化和主从复制时不会阻塞其他操作，提高了整体的性能。
    优化的网络协议：Redis使用RESP（Redis Serialization Protocol）作为网络协议，RESP是一种简单、高效的二进制协议。RESP协议的设计使得网络传输的数据量尽可能地减少，减少了网络传输的开销，提高了性能。
  • 原子性操作：Redis支持很多原子性操作，比如INCR、DECR、SETNX等。原子性操作能够在一条命令中完成多个操作，而且这些操作是不可中断的，保证了数据的一致性。

应用场景

由于 Redis 具备高性能、丰富的数据结构和多种特性，它的主要应用场景如下：

  • 缓存：作为缓存数据库，Redis 可以将经常访问的数据存储在内存中，避免频繁读写数据库，提高应用的响应速度。缓存可以存储热点数据，减轻后端数据库的压力，提高系统的吞吐量。
  • 计数器：利用 Redis 的原子性操作，可以实现高效的计数器功能，比如网站的点赞、浏览次数等统计功能。由于 Redis 原子性操作的特性，计数器的更新可以并发执行而不会出现竞争问题。
  • 消息队列：Redis 的发布订阅功能和列表数据结构可以实现简单的消息队列，用于解耦系统的各个模块。生产者将消息发布到指定频道，消费者订阅感兴趣的频道并处理消息，实现异步消息传递。
    排行榜：使用有序集合数据结构，可以实现排行榜功能，比如游戏中的玩家排名。通过有序集合的分数属性，可以对玩家的得分进行排序和排名，实时显示排行榜。
  • 会话缓存：在 Web 应用中，可以使用 Redis 存储用户的会话数据，实现分布式会话管理。用户登录后，可以将会话数据存储在 Redis 中，从而实现多台服务器之间的会话共享。

Redis 常用命令

Redis 提供了命令行客户端 redis-cli，用于与 Redis 服务进行交互。以下是一些常用的 Redis 命令：

SET：添加或修改一个字符串类型的键值对。
GET：根据键名获取对应的值。
INCR：将一个整数值自增 1。
DECR：将一个整数值自减 1。
LPUSH：向列表的左边插入一个元素。
RPUSH：向列表的右边插入一个元素。
HSET：添加或修改一个哈希类型键的字段值。
HGET：根据键名和字段名获取哈希类型的值。

Python 中使用 Redis

安装 redis-py 库

1
pip install redis

连接 Redis 服务器

1
2
3
4
5
6
7
8
9
10
import redis  
  
# 连接到本地 Redis 服务器  
r = redis.Redis(host='localhost', port=6379, db=0, password='', decode_responses=True)  
  
# host ip地址 默认本地
# port 端口号 默认6379
# password 密码
# db 数据库
# decode_responses=True 参数用于确保从 Redis 获取的字符串数据以 Python 字符串形式存在，而不是字节类型

数据操作

字符串操作

  • key 代指键名
  • string 代指值（字符串类型）

设置键值对 set(key, string)

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

获取字符串 get(key)

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

检查键是否存在 exists(key)

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

删除键 delete(key)

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

批量设置键值对 mset({key: value, key1: value1})

1
2
3
r.mset({'key1': 'value1', 'key2': 'value2'})  
values = r.mget('key1', 'key2')  
print(values)  # 输出: ['value1', 'value2']

哈希操作

  • keytable 代指哈希表
  • key 字段
  • value 数据

存储哈希值hset(keytable, key, value)

1
r.hset('myhash', 'field1', 'value1')

获取哈希值hget(keytable, key)

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

批量存储哈希值hget(keytable, key_dict)

1
2
3
4
# 假设 hmset 可用  
r.hmset('myhash', {'field2': 'value2', 'field3': 'value3'})  
values = r.hmget('myhash', 'field2', 'field3')  
print(values)  # 输出: [b'value2', b'value3'] （注意：这里可能需要 decode_responses=True）

获取所有哈希值hgetall(keytable)

1
2
all_fields_values = r.hgetall('myhash')  
print(all_fields_values)  # 输出类似: {'field1': 'value1', 'field2': 'value2', ...}

列表操作

  • mylist 代指列表名
  • value代指数据

添加元素

  • lpush(mylist, value) 在列表左侧添加
  • rpush(mylist, value) 在列表右侧添加
1
2
r.lpush('mylist', 'value1')  # 在列表左侧添加  
r.rpush('mylist', 'value2')  # 在列表右侧添加

获取列表元素lrang(mylist, start, end)

  • start 索引开始位置
  • end 索引结束位置
1
2
3
4
5
# 获取列表中的所有元素  
print(r.lrange('mylist', 0, -1))  # 输出: [b'0', b'a', b'b', b'c']  
  
# 获取列表中的前两个元素  
print(r.lrange('mylist', 0, 1))  # 输出: [b'0', b'a']

移除元素

  • lpop：移除并获取列表的第一个元素
  • rpop：移除并获取列表的最后一个元素
1
2
3
4
5
# 移除并获取列表的第一个元素  
print(r.lpop('mylist'))  # 输出: b'0'  
  
# 移除并获取列表的最后一个元素  
print(r.rpop('mylist'))  # 输出: b'c'

列表长度llen(mylist)

1
print(r.llen('mylist'))  # 输出列表的长度

集合操作

  • myset 集合名
  • value 数据

添加元素sadd(myset, value)

1
2
r.sadd('myset', 'element1')  
r.sadd('myset', 'element2', 'element3')  # 可以一次添加多个元素

获取集合中的元素smember(myset)

1
2
elements = r.smembers('myset')  
print(elements)  # 输出集合中的所有元素，可能是一个无序的列表

集合的长度scart(myset)

1
2
length = r.scard('myset')  
print(length)  # 输出集合中的元素数量

集合的差集sdiff(myset, myset1)

1
2
3
4
5
6
set1 = {'element1', 'element2', 'element3'}  
r.sadd('set1', *set1)  # 假设set1已存在于Redis中  
r.sadd('set2', 'element2', 'element4')  # 假设set2也已存在  
  
diff = r.sdiff('set1', 'set2')  
print(diff)  # 输出: {'element1', 'element3'}，即set1有而set2没有的元素

集合的交集sinter(myset, myset1)

1
2
inter = r.sinter('set1', 'set2')  
print(inter)  # 输出: {'element2'}，即set1和set2共有的元素

从集合中移除元素
使用srem命令从集合中移除一个或多个元素。

1
2
r.srem('myset', 'element1', 'element2')  
# 现在'myset'中不再包含'element1'和'element2'

集合的随机元素
使用srandmember命令从集合中随机获取一个或多个元素。

1
2
3
4
5
6
random_element = r.srandmember('myset')  
print(random_element)  # 随机输出'myset'中的一个元素  
  
# 获取多个随机元素  
random_elements = r.srandmember('myset', 2)  # 获取两个随机元素  
print(random_elements)

集合的弹出元素
使用spop命令从集合中随机移除一个元素并返回该元素。

1
2
popped_element = r.spop('myset')  
print(popped_element)  # 随机输出'myset'中的一个元素，并从集合中移除它

有序集合

有序集合是Redis中一种特殊的集合类型，它不仅包含元素，还为每个元素关联了一个浮点数分数（score），这使得元素可以按照分数进行排序。

添加元素到有序集合
使用zadd命令向有序集合中添加一个或多个元素及其分数。

1
2
3
r.zadd('mysortedset', {'member1': 1, 'member2': 2})  
# 或者对于单个元素  
r.zadd('mysortedset', 1, 'member3')

获取有序集合中的元素
使用zrange（升序）或zrevrange（降序）命令获取有序集合中的元素及其分数。

1
2
3
4
5
6
7
# 升序获取  
ascending = r.zrange('mysortedset', 0, -1, withscores=True)  
print(ascending)  # 输出类似: [('member1', 1.0), ('member2', 2.0), ('member3', 1.0)]  
  
# 降序获取  
descending = r.zrevrange('mysortedset', 0, -1, withscores=True)  
print(descending)  # 输出可能反转的顺序

获取有序集合的排名
使用zrank（升序排名）或zrevrank（降序排名）命令获取元素的排名。

1
2
3
4
5
rank = r.zrank('mysortedset', 'member1')  
print(rank)  # 输出: 0，如果member1是分数最低的元素  
  
revrank = r.zrevrank('mysortedset', 'member2')  
print(revrank)  # 输出: 0，如果member2是分数最高的元素

删除有序集合中的元素
使用zrem命令从有序集合中移除一个或多个元素。

1
r.zrem('mysortedset', 'member1', 'member3')

使用Redis作为缓存

为了避免短时间内大量的请求访问导致读取数据频繁，可以使用Redis作为缓存来提高API接口的响应速度。我的思路是将函数名和参数作为key，返回值作为value，并根据数据的变动情况设置缓存过期时间。基本不变的数据可以设置较长的过期时间，而经常变动的数据可以设置较短的过期时间，既能提高缓存命中率，又能避免缓存的滞后性。

同时，为了方便存入和取出数据，我将缓存函数写为了装饰器(decorator)，可以直接在函数上添加@cache()装饰器，即可在调用函数前先检查缓存，如果缓存存在，则直接返回缓存数据，否则调用函数并将结果存入缓存。

另外，我还在装饰器中添加了refresh参数的识别，如果函数传参时带有refresh=True参数，装饰器会自动将其剔除，不会影响原函数执行，但会强制刷新缓存，即不检查缓存是否存在，这样可以更灵活地控制缓存的更新，比如某些特殊情况下需要即时更新缓存的函数。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
import redis
from functools import wraps

# 此处是自定义的配置文件，用于存放Redis连接信息等
# REDIS_HOST redis主机地址
# REDIS_PORT redis端口
# REDIS_DB redis数据库
# EXPIRE_TIME 缓存过期时间，单位秒
# CACHE_PREFIX 缓存前缀，用于区分同一数据库下不同项目的缓存
# ENABLE_CACHE 是否启用缓存
from config import REDIS_HOST, REDIS_PORT, REDIS_DB, EXPIRE_TIME, CACHE_PREFIX, ENABLE_CACHE

# 建立Redis连接
# 由于我的redis未设置密码，所以这里无需连接密码，如果设置了密码，则需要修改此处代码
redis_db = redis.Redis(host=REDIS_HOST, port=REDIS_PORT, db=REDIS_DB, decode_responses=True)


def cache(cache_exp=EXPIRE_TIME):
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            # 提取刷新标志，并从kwargs中移除，以防止传递给原始函数
            refresh = kwargs.pop('refresh', False)
            # 使用函数名称和参数生成唯一的Redis键
            cache_key = f"{CACHE_PREFIX}:{func.__name__}:{args}"

            # 如果缓存未启用，直接调用原始函数
            if not ENABLE_CACHE:
                return func(*args, **kwargs)

            # 检查是否需要刷新缓存
            if not refresh:
                try:
                    # 查询Redis缓存
                    cached_result = redis_db.get(cache_key)
                    if cached_result:
                        return eval(cached_result)
                except:
                    pass

            # 如果缓存不存在或已过期，调用原始函数
            result = func(*args, **kwargs)
            # 将结果存储到Redis，并设置过期时间
            if result != {} and result != '{}':
                redis_db.setex(cache_key, cache_exp, str(result))
            else:
                redis_db.setex(cache_key, 180, str(result))
            return result

        return wrapper
    return decorator

使用方法：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
# 调用缓存装饰器，过期时间为默认时间
@cache()
def my_func(a, b):
    # 此处是需要缓存的函数
    return a + b

# 调用缓存装饰器，过期时间为120秒
@cache(120)
def my_func1(a, b):
    # 此处是需要缓存的函数
    return a + b

# 调用函数，并传入参数
result = my_func(1, 2)
print(result)  # 输出: 3

# 再次调用函数，由于缓存存在，所以直接返回缓存数据，不执行函数
result = my_func(1, 2)
print(result)  # 输出: 3

# 调用带有refresh参数的函数，会强制刷新缓存，重新执行函数并更新缓存
result = my_func(1, 2, refresh=True)
print(result)  # 输出: 3