Redis缓存数据库实战教程

本文集文档发布于灏天文库

3. Redis 高级特性

文档摘要

Redis 高级特性 Redis 高级特性详解与实践 本文将重点介绍以下 Redis 高级特性： 发布/订阅 (Pub/Sub)：构建实时消息传递系统。 事务 (Transactions)：保证操作的原子性。 Lua 脚本 (Lua Scripting)：扩展 Redis 功能，实现复杂逻辑。 管道 (Pipelines)：优化客户端与服务器的交互，提升性能。 持久化 (Persistence)：保障数据安全，防止数据丢失。 集群 (Cluster)：构建高可用、高扩展性的 Redis 服务。 请注意： 本文的代码示例将主要使用 Python 语言和 客户端库进行演示，但概念和原理适用于各种 Redis 客户端。

3. Redis 高级特性

Redis 高级特性详解与实践

本文将重点介绍以下 Redis 高级特性：

  1. 发布/订阅 (Pub/Sub)：构建实时消息传递系统。

  2. 事务 (Transactions)：保证操作的原子性。

  3. Lua 脚本 (Lua Scripting)：扩展 Redis 功能，实现复杂逻辑。

  4. 管道 (Pipelines)：优化客户端与服务器的交互，提升性能。

  5. 持久化 (Persistence)：保障数据安全，防止数据丢失。

  6. 集群 (Cluster)：构建高可用、高扩展性的 Redis 服务。

请注意： 本文的代码示例将主要使用 Python 语言和 redis-py 客户端库进行演示，但概念和原理适用于各种 Redis 客户端。

1. 发布/订阅 (Pub/Sub)

概念详解：

Redis 的发布/订阅模式允许消息的发送者（Publisher）将消息发布到特定的频道（Channel），而订阅者（Subscriber）可以订阅一个或多个频道，接收所有发布到这些频道的消息。这是一种典型的消息队列模式，但 Redis Pub/Sub 更侧重于实时消息的广播，而非消息的持久化和可靠传递（与专业的消息队列系统如 RabbitMQ 或 Kafka 不同）。

核心命令：

  • PUBLISH channel message: 将 message 发布到指定的 channel。

  • SUBSCRIBE channel [channel ...]: 订阅一个或多个频道。

  • PSUBSCRIBE pattern [pattern ...]: 根据模式订阅频道，可以使用通配符 * 和 ?。

  • UNSUBSCRIBE [channel [channel ...]]: 取消订阅指定的频道，如果不指定频道则取消所有订阅。

  • PUNSUBSCRIBE [pattern [pattern ...]]: 取消订阅符合模式的频道。

  • PUBSUB CHANNELS [pattern]: 列出活跃的频道，可以根据模式进行过滤。

  • PUBSUB NUMSUB [channel [channel ...]]: 获取指定频道的订阅者数量。

  • PUBSUB NUMPAT: 获取模式订阅的数量。

代码实践 (Python - redis-py)：

import redis
import time
import threading
# 连接 Redis
r = redis.Redis(host='localhost', port=6379, db=0)
# 订阅者函数
def subscriber_handler(r, channels):
    pubsub = r.pubsub()
    pubsub.subscribe(channels)  # 订阅频道
    for message in pubsub.listen():  # 监听消息
        if message['type'] == 'message':
            channel = message['data'].decode('utf-8')
            data = message['data'].decode('utf-8')
            print(f"Subscriber received message: {data} from channel: {channel}")
# 发布者函数
def publisher(r, channel, message):
    r.publish(channel, message)
    print(f"Publisher sent message: {message} to channel: {channel}")
if __name__ == "__main__":
    channel_name = "my_channel"
    # 启动订阅者线程
    subscriber_thread = threading.Thread(target=subscriber_handler, args=(r, [channel_name]))
    subscriber_thread.start()
    time.sleep(1)  # 等待订阅者启动
    # 发布消息
    publisher(r, channel_name, "Hello from publisher 1")
    publisher(r, channel_name, "Another message for subscribers")
    # 保持主线程运行一段时间，以便接收更多消息 (实际应用中订阅者会持续运行)
    time.sleep(3)
​

代码详解：

  1. 连接 Redis: 使用 redis.Redis() 创建 Redis 连接实例。

  2. 订阅者函数 subscriber_handler:

    • 使用 r.pubsub() 创建 Pub/Sub 对象。

    • pubsub.subscribe(channels) 订阅指定的频道列表。

    • pubsub.listen() 进入消息监听循环，这是一个迭代器，会持续接收新的消息。

    • 遍历 pubsub.listen() 返回的消息，判断消息类型是否为 'message' (普通消息)。

    • 从消息中提取频道名和数据，并打印出来。

  3. 发布者函数 publisher:

    • 使用 r.publish(channel, message) 将消息发布到指定频道。

  4. 主程序 if __name__ == "__main__"::

    • 定义频道名 channel_name。

    • 创建并启动订阅者线程，将 subscriber_handler 函数作为线程目标，并传入 Redis 连接和频道名。

    • 使用 time.sleep(1) 稍微等待订阅者线程启动并完成订阅。

    • 调用 publisher 函数发布两条消息到 channel_name 频道。

    • 使用 time.sleep(3) 保持主线程运行一段时间，以便订阅者接收并处理消息。

应用场景：

  • 实时聊天系统: 用户之间的消息可以发布到频道，所有在线用户订阅相关频道即可接收消息。

  • 实时通知: 系统事件（如订单状态更新、服务告警）可以发布到频道，相关服务或监控系统订阅频道即可接收通知。

  • 配置中心: 配置变更可以发布到频道，应用程序订阅频道即可实时更新配置。

  • 服务器之间的命令广播: 在分布式系统中，可以使用 Pub/Sub 进行命令广播，例如通知集群中的其他节点进行某些操作。

注意事项：

  • 消息无持久化： Redis Pub/Sub 不会持久化消息。如果订阅者离线，在离线期间发布的消息将会丢失。

  • 消息的广播性： 消息会被广播到所有订阅了该频道的订阅者，每个订阅者都会收到相同的消息。

  • 模式订阅的性能： 如果模式订阅数量过多，可能会对 Redis 服务器的性能产生一定影响，因为服务器需要遍历所有模式来匹配频道。

2. 事务 (Transactions)

概念详解：

Redis 事务允许将多个命令打包成一个原子操作序列。事务中的所有命令要么全部执行成功，要么全部不执行。Redis 事务提供了 原子性 (Atomicity)、隔离性 (Isolation)，但 不保证回滚 (Rollback)。 在 Redis 事务中，如果一个命令在入队时语法错误，Redis 会拒绝执行整个事务。如果在执行阶段出现错误（例如，操作了错误的数据类型），Redis 不会回滚已执行的命令，而是会继续执行事务中剩余的命令。

核心命令：

  • MULTI: 标记事务块的开始。在此命令之后的所有命令都将被放入队列中，直到 EXEC 命令执行。

  • EXEC: 执行事务队列中的所有命令。

  • DISCARD: 清空事务队列，放弃事务。

  • WATCH key [key ...]: 监视一个或多个键。如果在事务执行 EXEC 命令之前，被监视的键被修改，事务将被取消。

  • UNWATCH: 取消所有 WATCH 命令的监视。

代码实践 (Python - redis-py)：

import redis
# 连接 Redis
r = redis.Redis(host='localhost', port=6379, db=0)
def execute_transaction(r):
    try:
        pipeline = r.pipeline()  # 创建 Pipeline 对象，Pipeline 继承自 Transaction
        pipeline.multi()        # 开始事务
        pipeline.incr('transaction_counter')  # 命令1：自增计数器
        pipeline.get('transaction_counter') # 命令2：获取计数器值
        result = pipeline.execute()      # 执行事务
        print(f"Transaction executed successfully. Results: {result}")
        return result
    except redis.exceptions.WatchError as e:
        print(f"Watch error occurred: {e}")
        return None
    except Exception as e:
        print(f"Transaction failed: {e}")
        return None
if __name__ == "__main__":
    r.set('transaction_counter', 0)  # 初始化计数器
    results = execute_transaction(r)
    if results:
        print(f"Counter value after transaction: {r.get('transaction_counter').decode()}")
    results2 = execute_transaction(r) # 再次执行事务
    if results2:
        print(f"Counter value after second transaction: {r.get('transaction_counter').decode()}")
​

代码详解：

  1. 连接 Redis: 同 Pub/Sub 示例。

  2. execute_transaction 函数:

    • pipeline = r.pipeline(): 创建 Pipeline 对象。在 redis-py 中，Pipeline 对象也用于实现事务。

    • pipeline.multi(): 显式地开始事务块。虽然在 Pipeline 中默认是原子操作，但显式 multi() 更清晰地表示事务的开始。

    • pipeline.incr('transaction_counter') 和 pipeline.get('transaction_counter'): 将 INCR 和 GET 命令添加到事务队列中。这些命令并不会立即执行。

    • result = pipeline.execute(): 执行事务队列中的所有命令。execute() 方法返回一个列表，列表中包含了每个命令的执行结果，顺序与命令入队顺序一致。

    • try...except 块用于捕获可能的异常，例如 redis.exceptions.WatchError (当使用 WATCH 监控的键被修改时) 和其他可能的异常。

  3. 主程序 if __name__ == "__main__"::

    • 初始化键 transaction_counter 的值为 0。

    • 调用 execute_transaction 函数执行事务，并打印事务执行结果和计数器的值。

    • 再次调用 execute_transaction 函数，演示事务的多次执行。

WATCH 乐观锁示例 (Python - redis-py)：

import redis
import time
import threading
# 连接 Redis
r = redis.Redis(host='localhost', port=6379, db=0)
def process_payment(r, order_id, amount):
    key = f"order:{order_id}:balance"
    while True:
        try:
            r.watch(key)  # 监视订单余额键
            balance = int(r.get(key) or 0) # 获取当前余额，如果不存在则为 0
            if balance >= amount:
                pipeline = r.pipeline()
                pipeline.multi()
                pipeline.decrby(key, amount) # 扣减余额
                pipeline.incr('payment_processed_count') # 增加支付处理计数器
                pipeline.execute() # 执行事务
                print(f"Order {order_id}: Payment of {amount} processed successfully.")
                return True
            else:
                r.unwatch() # 取消监视
                print(f"Order {order_id}: Insufficient balance. Current balance: {balance}, required: {amount}")
                return False
        except redis.exceptions.WatchError:
            print(f"Order {order_id}: Conflict detected. Retrying...")
            time.sleep(0.1) # 短暂等待后重试
if __name__ == "__main__":
    order_id = "12345"
    r.set(f"order:{order_id}:balance", 100) # 初始化订单余额为 100
    r.set('payment_processed_count', 0) # 初始化支付处理计数器
    # 模拟并发支付请求
    threads = []
    for i in range(3):
        thread = threading.Thread(target=process_payment, args=(r, order_id, 30)) # 每个线程尝试支付 30
        threads.append(thread)
        thread.start()
    for thread in threads:
        thread.join()
    print(f"Final order balance: {r.get(f'order:{order_id}:balance').decode()}")
    print(f"Total payments processed: {r.get('payment_processed_count').decode()}")
​

代码详解 (乐观锁示例)：

  1. process_payment 函数:

    • 使用 r.watch(key) 监视订单余额键 key。

    • 在 while True 循环中不断尝试支付，直到成功或余额不足。

    • 获取当前余额 balance。

    • 检查余额是否足够支付。

    • 如果余额足够：

      • 创建 Pipeline 对象并开始事务。

      • 使用 pipeline.decrby(key, amount) 扣减余额。

      • 使用 pipeline.incr('payment_processed_count') 增加支付处理计数器。

      • pipeline.execute() 执行事务。

      • 支付成功，返回 True。

    • 如果余额不足：

      • r.unwatch() 取消监视。

      • 支付失败，返回 False。

    • 如果捕获到 redis.exceptions.WatchError 异常，表示在事务执行期间被监视的键被修改（并发冲突），则打印冲突信息，短暂等待后重试事务。

  2. 主程序 if __name__ == "__main__"::

    • 初始化订单余额和支付处理计数器。

    • 创建 3 个线程，每个线程尝试支付 30。

    • 启动并等待所有线程完成。

    • 打印最终订单余额和支付处理计数器值。

应用场景：

  • 电商系统库存扣减: 使用 WATCH 实现乐观锁，防止超卖。

  • 银行转账: 保证转账操作的原子性，要么同时扣款和收款成功，要么都失败。

  • 计数器更新: 在并发环境下，使用事务保证计数器更新的原子性。

注意事项：

  • Redis 事务不是 ACID 事务： Redis 事务只能保证原子性和隔离性，但不保证持久性 (Durability) 和一致性 (Consistency) 的完整 ACID 属性。

  • 没有回滚： Redis 事务不支持回滚。如果事务执行过程中出现错误，已执行的命令不会被撤销。

  • 乐观锁的应用场景： WATCH 命令适用于乐观锁场景，即假设并发冲突的概率较低，先执行操作，如果发现冲突则重试。

3. Lua 脚本 (Lua Scripting)

概念详解：

Redis 允许执行 Lua 脚本。Lua 脚本可以在 Redis 服务器端原子地执行多个 Redis 命令。使用 Lua 脚本可以：

  • 原子性操作： 保证脚本中的多个命令作为一个原子操作执行，避免竞态条件。

  • 减少网络开销： 将多个操作放在一个脚本中执行，减少客户端与服务器之间的网络交互次数，提升性能。

  • 扩展 Redis 功能： Lua 脚本可以实现复杂的业务逻辑，扩展 Redis 的功能。

核心命令：

  • EVAL script numkeys key [key ...] arg [arg ...]: 执行 Lua 脚本。script 是 Lua 脚本内容，numkeys 是键参数的数量，key [key ...] 是键参数列表，arg [arg ...] 是其他参数列表。键参数和普通参数在 Lua 脚本中可以通过 KEYS 和 ARGV 数组访问。

  • EVALSHA sha1 numkeys key [key ...] arg [arg ...]: 执行已加载到 Redis 服务器的 Lua 脚本。sha1 是脚本内容的 SHA1 摘要。

  • SCRIPT LOAD script: 将 Lua 脚本加载到 Redis 服务器，返回脚本的 SHA1 摘要。

  • SCRIPT EXISTS sha1 [sha1 ...]: 检查指定的 SHA1 摘要的脚本是否存在于 Redis 服务器。

  • SCRIPT FLUSH: 从脚本缓存中移除所有脚本。

  • SCRIPT KILL: 终止当前正在执行的脚本。

代码实践 (Python - redis-py)：

import redis
# 连接 Redis
r = redis.Redis(host='localhost', port=6379, db=0)
def execute_lua_script(r):
    lua_script = """
    local counter_key = KEYS[1]
    local increment_value = tonumber(ARGV[1])
    local current_value = redis.call('GET', counter_key)
    if not current_value then
        current_value = 0
    end
    current_value = tonumber(current_value) + increment_value
    redis.call('SET', counter_key, current_value)
    return current_value
    """
    result = r.eval(lua_script, 1, 'my_lua_counter', 5) # 执行 Lua 脚本，1个键参数，键名为 'my_lua_counter'，参数值为 5
    print(f"Lua script executed. Result: {result}")
    return result
def execute_lua_script_sha(r, sha1):
    result = r.evalsha(sha1, 1, 'my_lua_counter', 3) # 使用 SHA1 执行已加载的脚本
    print(f"Lua script (SHA) executed. Result: {result}")
    return result
if __name__ == "__main__":
    r.set('my_lua_counter', 10) # 初始化计数器
    result1 = execute_lua_script(r) # 执行 Lua 脚本
    print(f"Counter value after script execution: {r.get('my_lua_counter').decode()}")
    # 加载 Lua 脚本到 Redis 服务器
    script_sha = r.script_load(execute_lua_script.__defaults__[0]) # 获取 execute_lua_script 函数默认的 lua_script
    print(f"Loaded Lua script SHA1: {script_sha}")
    result2 = execute_lua_script_sha(r, script_sha) # 使用 SHA1 执行脚本
    print(f"Counter value after SHA script execution: {r.get('my_lua_counter').decode()}")
​

代码详解：

  1. 连接 Redis: 同 Pub/Sub 示例。

  2. execute_lua_script 函数:

    • lua_script = """...""": 定义 Lua 脚本字符串。

    • Lua 脚本内容详解:

      • local counter_key = KEYS[1]: 获取第一个键参数 (在 Python 代码中传入的 'my_lua_counter')。

      • local increment_value = tonumber(ARGV[1]): 获取第一个参数 (在 Python 代码中传入的 5)，并转换为数字类型。

      • local current_value = redis.call('GET', counter_key): 使用 redis.call 调用 Redis 命令 GET 获取计数器当前值。

      • if not current_value then current_value = 0 end: 如果计数器不存在，则初始化为 0。

      • current_value = tonumber(current_value) + increment_value: 将当前值转换为数字，加上增量值。

      • redis.call('SET', counter_key, current_value): 使用 redis.call 调用 Redis 命令 SET 更新计数器值。

      • return current_value: 返回更新后的计数器值。

    • result = r.eval(lua_script, 1, 'my_lua_counter', 5): 使用 r.eval 执行 Lua 脚本。

      • 第一个参数 lua_script 是脚本内容。

      • 第二个参数 1 表示键参数的数量为 1。

      • 第三个参数 'my_lua_counter' 是第一个键参数，对应 Lua 脚本中的 KEYS[1]。

      • 第四个参数 5 是第一个普通参数，对应 Lua 脚本中的 ARGV[1]。

  3. execute_lua_script_sha 函数:

    • result = r.evalsha(sha1, 1, 'my_lua_counter', 3): 使用 r.evalsha 执行已加载的 Lua 脚本，参数与 r.eval 类似，只是将脚本内容替换为 SHA1 摘要。

  4. 主程序 if __name__ == "__main__"::

    • 初始化计数器 my_lua_counter 为 10。

    • 调用 execute_lua_script 执行 Lua 脚本，并打印结果和计数器值。

    • script_sha = r.script_load(execute_lua_script.__defaults__[0]): 使用 r.script_load 将 Lua 脚本加载到 Redis 服务器，并获取脚本的 SHA1 摘要。execute_lua_script.__defaults__[0] 获取的是 execute_lua_script 函数默认参数中的 lua_script 字符串。

    • 调用 execute_lua_script_sha 使用 SHA1 摘要执行脚本，并打印结果和计数器值。

应用场景：

  • 原子性操作复杂逻辑： 例如，原子性地检查并更新多个键的值，实现更复杂的业务逻辑。

  • 限流： 使用 Lua 脚本实现原子性的访问频率控制。

  • 排行榜更新： 原子性地更新排行榜数据。

  • 自定义 Redis 命令： 通过 Lua 脚本扩展 Redis 的功能，实现自定义的命令。

注意事项：

  • 性能影响： 长时间运行的 Lua 脚本会阻塞 Redis 服务器，影响性能。应尽量编写执行时间短的脚本。

  • 脚本调试： Lua 脚本在 Redis 服务器端执行，调试相对复杂。可以使用 redis-cli --eval 命令进行调试。

  • 脚本缓存： 使用 SCRIPT LOAD 和 EVALSHA 可以将脚本预加载到 Redis 服务器，提高执行效率，并减少网络传输。

4. 管道 (Pipelines)

概念详解：

Redis 管道 (Pipelines) 是一种批量执行 Redis 命令的机制。在传统的客户端-服务器交互模式中，客户端发送一个命令，等待服务器响应，然后再发送下一个命令。使用管道，客户端可以将多个命令一次性发送给服务器，服务器在接收到所有命令后，一次性处理所有命令，并将所有结果一次性返回给客户端。这样可以显著减少客户端与服务器之间的网络往返次数 (Round Trip Time, RTT)，从而提升性能，尤其是在需要执行大量命令时。

代码实践 (Python - redis-py)：

import redis
import time
# 连接 Redis
r = redis.Redis(host='localhost', port=6379, db=0)
def execute_without_pipeline(r, num_commands):
    start_time = time.time()
    for i in range(num_commands):
        r.set(f"key:{i}", i)
    end_time = time.time()
    duration = end_time - start_time
    print(f"Executed {num_commands} commands without pipeline in {duration:.4f} seconds.")
    return duration
def execute_with_pipeline(r, num_commands):
    start_time = time.time()
    pipeline = r.pipeline() # 创建 Pipeline 对象
    for i in range(num_commands):
        pipeline.set(f"key:{i}", i) # 将 SET 命令添加到管道
    pipeline.execute() # 执行管道中的所有命令
    end_time = time.time()
    duration = end_time - start_time
    print(f"Executed {num_commands} commands with pipeline in {duration:.4f} seconds.")
    return duration
if __name__ == "__main__":
    num_commands = 10000
    duration_without_pipeline = execute_without_pipeline(r, num_commands)
    duration_with_pipeline = execute_with_pipeline(r, num_commands)
    print(f"Pipeline speedup factor: {duration_without_pipeline / duration_with_pipeline:.2f}x")
​

代码详解：

  1. 连接 Redis: 同 Pub/Sub 示例。

  2. execute_without_pipeline 函数:

    • 循环 num_commands 次，每次调用 r.set() 单独执行 SET 命令。

    • 记录开始时间和结束时间，计算执行时间。

  3. execute_with_pipeline 函数:

    • pipeline = r.pipeline(): 创建 Pipeline 对象。

    • 循环 num_commands 次，每次调用 pipeline.set() 将 SET 命令添加到管道中。这些命令并没有立即执行，而是被放入管道队列中。

    • pipeline.execute(): 执行管道中的所有命令。Redis 服务器一次性处理管道中的所有命令，并将结果返回给客户端。

    • 记录开始时间和结束时间，计算执行时间。

  4. 主程序 if __name__ == "__main__"::

    • 设置命令数量 num_commands 为 10000。

    • 分别调用 execute_without_pipeline 和 execute_with_pipeline 函数，测试不使用管道和使用管道的性能。

    • 计算并打印管道加速因子，即不使用管道的执行时间与使用管道的执行时间的比值。

应用场景：

  • 批量数据操作： 例如，批量导入数据、批量更新数据、批量删除数据等。

  • 需要执行大量命令的场景： 例如，初始化缓存、执行复杂的 ETL 任务等。

注意事项：

  • 原子性： Pipeline 中的命令不是原子性执行的，与事务不同。如果 Pipeline 执行过程中出现错误，之前的命令可能已经执行成功。

  • 性能提升： 管道主要通过减少网络 RTT 来提升性能，对于 CPU 密集型操作，管道的性能提升可能不明显。

  • 命令类型： Pipeline 可以用于执行任何 Redis 命令，不仅仅是 SET 命令。

5. 持久化 (Persistence)

概念详解：

Redis 是一种内存数据库，数据存储在内存中，速度非常快。但是，内存中的数据是非持久化的，一旦 Redis 服务器重启或宕机，内存中的数据将会丢失。为了保证数据的安全性，Redis 提供了两种持久化机制：

  • RDB (Redis DataBase): 快照持久化。RDB 会定期将 Redis 在内存中的数据集快照写入磁盘上的 RDB 文件。

  • AOF (Append Only File): 追加文件持久化。AOF 会将 Redis 服务器接收到的每个写命令追加到 AOF 文件的末尾。在 Redis 服务器重启时，会重新执行 AOF 文件中的所有命令来恢复数据。

RDB 持久化：

  • 优点：

    • 性能高： RDB 持久化是后台进程 fork 子进程来完成的，主进程不需要进行任何 IO 操作，对性能影响较小。

    • 恢复速度快： RDB 文件是数据集的快照，恢复数据时只需要加载 RDB 文件即可，恢复速度比 AOF 快。

  • 缺点：

    • 数据丢失风险： RDB 是定期快照，如果在两次快照之间 Redis 服务器宕机，会丢失这段时间的数据。

    • fork 开销： 如果数据集很大，fork 子进程会比较耗时，可能会导致 Redis 服务器短暂的停顿。

AOF 持久化：

  • 优点：

    • 数据安全性高： AOF 可以配置不同的 fsync 策略 (always, everysec, no)，always 策略下，每次写命令都会同步到磁盘，数据安全性最高，数据丢失风险最低。

    • 可读性高： AOF 文件是文本文件，记录了 Redis 的写命令，可读性较高，可以用于数据恢复和审计。

  • 缺点：

    • 性能略低： AOF 持久化需要进行 IO 操作，性能比 RDB 略低，尤其是在 always 策略下。

    • 文件体积较大： AOF 文件会记录所有写命令，文件体积通常比 RDB 文件大。

    • 恢复速度较慢： 恢复数据时需要重新执行 AOF 文件中的所有命令，恢复速度比 RDB 慢。

作者与出处
原作者： 灏天文库
来源：灏天文库
整理： 灏天文库整理
由灏天文库平台收录，内容或由平台用户上传，仅供学习交流
发布者: 作者: 灏天文库 转发
评论区 (0)
U
0/500