Python3使用python-redis-lock解决Redis高并发读写冲突

Evan

发布时间：2026-10-14 | 4人浏览过

原创

本文针对Redis高并发场景下客户端读写数据相互覆盖的问题，介绍使用python-redis-lock库实现分布式锁的方法。通过示例代码演示如何获取锁、执行业务逻辑并释放锁，有效避免多客户端同时读取同一数据导致的计算错误，确保数据一致性。

需求

在最近的一项任务中，涉及Redis高并发场景下计算多个客户端接收预警信息的时长问题。

系统模型首先模拟多个客户端连接至预警服务器集群，并向集群发送告警信息。随后，预警服务集群会向这些客户端推送告警通知。

在此过程中，系统记录从发送告警到预警集群的时间，同时客户端也记录接收到告警的具体时间。

上述时间数据均被存储至Redis中。此时面临一个关键问题：当多个客户端并发地执行读取、计算和设置数据的操作时，这些操作会相互覆盖，导致数据不一致。

从相关截图可以看出，多个不同的客户端在并发读取Redis数据时，往往读取到相同的旧数据，进而导致后续计算出现错误。

导致该问题的具体流程示意图如下：

为解决此问题，需要引入Redis锁机制来控制数据的并发读取与写入。Python原生Redis库仅支持乐观锁的一种写法，此处推荐使用python-redis-lock库。该库能方便地对Redis多客户端并发场景加锁，有效处理竞争条件。下面将展示具体使用方法。

python-redis-lock

项目官方地址：https://pypi.org/project/python-redis-lock/

在使用该库之前，需先进行安装：


pip install python-redis-lock

以下是使用锁的基本示例：


lock = redis_lock.Lock(conn, "name-of-the-lock")
if lock.acquire(blocking=False):
    print("Got the lock.")
    lock.release()
else:
    print("Someone else has the lock.")

上述示例展示了单独设置锁的方式，此外还可以将所有Redis操作封装在锁中执行。


# On application start/restart
import redis_lock
redis_lock.reset_all(redis_client)

修改业务代码，增加lock操作

1. 首先导入redis_lock模块


import redis_lock

2. 将Redis连接的客户端对象传入Lock中，并指定锁的名称


# 设置redis连接
self.conn = redis.Redis(host='127.0.0.1', port=6379, decode_responses=True, db=3)
 
# 设置redis锁
self.lock = redis_lock.Lock(self.conn, "redis-lock")

3. 将业务逻辑中读取和设置Redis的部分代码包裹在锁操作内


while True:
    # 设置redis锁，操作redis
    if self.lock.acquire(blocking=False):
        print("Got the lock.")
        # 获取lock，执行业务处理
        # 获取当前redis钟记录的客户端接收到告警的总时长
        recv_time_sum_count_clients = self.conn.get(recv_time_sum_count_clients_key)
        if recv_time_sum_count_clients is None:
            recv_time_sum_count_clients = "0:0"
 
        # 获取当前的统计数据
        recv_time_sum, count_clients = recv_time_sum_count_clients.split(":")
 
        # 计算告警接收总时长
        recv_time_sum = float(recv_time_sum) + recv_time
        # 计算收到预警的客户端数量
        count_clients = int(count_clients) + 1
 
        # 写入redis中
        recv_time_sum_count_clients = "%s:%s" % (str(recv_time_sum), str(count_clients))
        self.conn.set(recv_time_sum_count_clients_key, recv_time_sum_count_clients)
 
        print("user_id = %s, 计算平均时间成功, "
              "recv_time_sum = %s, count_clients = %s \n" %
              (self.user_id, recv_time_sum, count_clients))
 
        # 释放lock
        self.lock.release()
 
        # 退出循环
        break
    else:
        print("Someone else has the lock.")

在客户端代码中设置锁后，再次执行测试，观察是否还存在抢占读取Redis数据的情况，结果如下：