Redis进阶（一）使用Redis实现分布式锁

本文将会介绍如何使用Redis来实现分布式锁，以及分布式锁的两个应用场景。

前言

在文章Redis快速入门，笔者介绍了Redis的基本数据结构和使用方法，可以作为初学者Redis入门文章。

笔者最近在使用定时任务的时候，发现一个问题：如果多台机器同时部署了同一个定时任务，则会出现同一资源被重复消费的问题。

解决的方案是使用分布式锁。那么，何为分布式锁？

分布式锁，顾名思义，就是在分布式环境下使用的锁，通过锁解决控制共享资源访问 的问题，来保证只有一个线程可以访问被保护的资源。用于实现分布式锁的组件通常都会具备以下的一些特性：

互斥性：提供分布式环境下的互斥，一个事件在同一个时间内只能被一个线程执行，这当然是分布式锁最基本的特性。
自动释放：为了应对分布式系统中各实例因通信故障导致锁不能释放的问题，自动释放的特性通常也是很有必要的。
分区容错性：应用在分布式系统的组件，具备分区容错性也是一项重要的特性，否则就会成为整个系统的瓶颈。

有多种方案可以实现分布式锁:

基于数据库实现
基于Zookeeper实现
基于Redis实现

本文将会重点介绍如何使用Redis来实现分布式锁。

分布式锁的使用场景列举如下：

防止缓存击穿 比如查询一个热点数据，如果缓存中没有，就会有大量请求同时去查数据库，加大数据库压力。加锁可以保证同一时间只有一个请求查数据库，其他请求等待或快速返回。
保证任务的唯一执行 比如定时任务在多个实例部署时，防止同一个任务被多个实例重复执行。加锁可以确保同一时刻，只有一个实例在执行。
控制并发资源 比如购买库存、发放优惠券、下单等高并发场景，需要保证库存扣减或资源分配的正确性，加锁能避免超卖或重复领取的问题。
跨系统、跨服务协调 在微服务架构中，不同服务之间可能需要协调某些操作，比如多个服务同时对一条数据进行修改，这时就需要分布式锁来保证一致性。
避免重复提交 用户短时间内重复提交相同的请求（比如支付按钮连续点击），可以通过加锁来防止后端执行两次。
主备切换场景 比如分布式系统中，某些服务需要通过抢占锁来决定谁是主节点（Leader Election）。

本文将会介绍分布式锁在保证任务的唯一执行和避免重复提交这两个场景中的应用。

Redis基础命令

在介绍如何使用Redis来实现分布式锁前，我们有必要先了解实现分布式锁的Redis基础命令。

SETNX命令

SETNX 是 set if not exists 的缩写，当且仅当 key 不存在时，则设置 value 给这个key。若给定的 key 已经存在，则 SETNX 不做任何动作。其返回值1表示该进程执行成功，将key值设置为value；返回值0表明该key已存在。

127.0.0.1:6379> SETNX demo_task 100
(integer) 1
127.0.0.1:6379> SETNX demo_task 100
(integer) 0

如上所示，第一次运行SETNX能成功，第二次运行时demo_task这个key已存在，返回值为0.

GET, DEL命令

GET命令获取key的值，如果存在，则返回；如果不存在，则返回nil.

DEL命令为删除对应key.

127.0.0.1:6379> GET demo_task
"100"
127.0.0.1:6379> DEL demo_task
(integer) 1
127.0.0.1:6379> GET demo_task
(nil)

设置超时

超时在分布式锁中就是重置，避免因为各种原因导致锁长时间无法释放，做法就是给key加个超时时间。

127.0.0.1:6379> SETNX demo_task 100
(integer) 1
127.0.0.1:6379> EXPIRE demo_task 3600
(integer) 1

上述命令设置demo_task这个key，并设置超时时间为3600秒。但上述操作分为两步执行，不是原子命令。

为了保证执行时的原子性，Redis 官方扩展了 SET 命令，既能满足获取对象，又能保证设置超时的时间语义。上述命令可以改写成：

127.0.0.1:6379> SET demo_task 100 NX PX 60000
OK
127.0.0.1:6379> GET demo_task
"100"

在上述命令中，NX表明Not Exist, PX表示超时时间，单位毫秒。

Python实现

基于上面的Redis基础命令，我们实现Python来实现分布式锁。需要注意的是，在释放锁（即运行删除命令）的时候，需要对锁进行唯一标识，避免别的程序误删除设置的锁。

下面是Python实现分布式锁的代码（文件名为distribute_key_op.py）:

# -*- coding: utf-8 -*-
import uuid
import math
import redis
from redis import WatchError


def acquire_lock_with_timeout(conn, lock_name, lock_timeout=2):
    """
    基于 Redis 实现的分布式锁

    :param conn: Redis 连接
    :param lock_name: 锁的名称
    :param lock_timeout: 锁的超时时间，默认 2 秒
    :return:
    """
    identifier = str(uuid.uuid4())
    lockname = f'lock:{lock_name}'
    lock_timeout = int(math.ceil(lock_timeout))

    # 如果不存在这个锁则加锁并设置过期时间，避免死锁
    if conn.set(lockname, identifier, ex=lock_timeout, nx=True):
        return identifier

    return False


def release_lock(conn, lockname, identifier):
    """
    释放锁

    :param conn: Redis 连接
    :param lockname: 锁的名称
    :param identifier: 锁的标识
    :return:
    """
    # python中redis事务是通过pipeline的封装实现的
    with conn.pipeline() as pipe:
        lockname = f'lock:{lockname}'

        while True:
            try:
                # watch 锁, 事务开始后如果该 key 被其他客户端改变, 事务操作会抛出 WatchError 异常
                pipe.watch(lockname)
                iden = pipe.get(lockname)
                if iden and iden.decode('utf-8') == identifier:
                    # 事务开始
                    pipe.multi()
                    pipe.delete(lockname)
                    pipe.execute()
                    return True

                pipe.unwatch()
                break
            except WatchError:
                print("WatchError")

        return False


if __name__ == '__main__':
    conn = redis.StrictRedis(host='localhost', port=6379, db=0)
    identifier = acquire_lock_with_timeout(conn, 'test', lock_timeout=120)
    print(f"identifier: {identifier}")
    result = release_lock(conn, 'test', identifier)
    print(f"result: {result}")

运行结果如下：

1 2	`identifier: e66ff09b-f26e-4067-b42b-898a2d1c46fa result: True`

定时任务唯一执行

设想场景：

在MySQL表，存在5条数据，需要设置定时任务，对doc字段进行关键词提取，并将结果保存为JSON文件。

MySQL表中对应数据如下：

该表使用sqlalchemy模块创建，对应Python代码（mysql_create_table.py）如下：

# -*- coding: utf-8 -*-
from sqlalchemy.dialects.mysql import INTEGER, TEXT
from sqlalchemy import Column
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
Base = declarative_base()


class Doc(Base):
    __tablename__ = 'user_doc'

    id = Column(INTEGER, primary_key=True, autoincrement=True)
    title = Column(TEXT, nullable=True)
    content = Column(TEXT, nullable=True)

    def __repr__(self):
        return f"Doc(id={self.id}, title={self.title}, content={self.content[:100]})"


def init_db():
    engine = create_engine(
        "mysql+pymysql://root:root@localhost:3306/orm_test",
        echo=True
    )
    # 创建表
    Base.metadata.create_all(engine)
    print('Create table successfully!')


if __name__ == '__main__':
    init_db()

定时任务(cron_doc_tag_1.py)脚本如下：

import os
import json
import asyncio
from dotenv import load_dotenv
from openai import AsyncOpenAI
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from apscheduler.schedulers.blocking import BlockingScheduler
from apscheduler.triggers.cron import CronTrigger

from mysql_create_table import Doc

load_dotenv()

# 初始化数据库连接
engine = create_engine("mysql+pymysql://root:root@localhost:3306/orm_test")
DBSession = sessionmaker(bind=engine)


async def process_doc(doc_id, doc_content):
    prompt = (
        "According to the content of the ducument, extract less than 10 key words "
        "from the content, and return them in a string separated by commas.\n"
        f"Document content: {doc_content}\n"
        "Key words:"
    )
    client = AsyncOpenAI(api_key=os.getenv("OPENAI_API_KEY"))
    response = await client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": prompt}]
    )
    result = response.choices[0].message.content
    with open("./redis_for_test/content_tags.json", "a") as f:
        json_data = {
            "doc_id": doc_id,
            "key_words": result,
            "task_id": "cron_doc_tag_1"
        }
        f.write(json.dumps(json_data, ensure_ascii=False) + "\n")
    print(f"doc_id: {doc_id}, key_words: {result}")


async def process_all_docs():
    # 创建新的会话
    session = DBSession()
    try:
        # 获取所有需要处理的文档
        docs = session.query(Doc).all()
        for doc in docs:
            await process_doc(doc.id, doc.content)
    finally:
        # 确保会话被关闭
        session.close()


def scheduled_job():
    """定时任务入口函数"""
    print("开始执行文档处理任务...")
    asyncio.run(process_all_docs())
    print("文档处理任务完成")


def main():
    """主函数：配置和启动调度器"""
    scheduler = BlockingScheduler()

    # 配置定时任务，每天20:30执行
    scheduler.add_job(
        scheduled_job,
        trigger=CronTrigger(hour=20, minute=30),
        id='process_docs_job',
        name='处理文档标签任务'
    )

    print("定时任务已启动，将在每天20:30执行...")
    try:
        scheduler.start()
    except (KeyboardInterrupt, SystemExit):
        print("定时任务已停止")


if __name__ == "__main__":
    main()

复制该脚本，重命名为cron_doc_tag_1.py，并将文件写入时"task_id"改为 "cron_doc_tag_2"。

同时运行上述两个脚本，则这两个定时任务在20点30分会同时运行，但是，数据库中的每条记录都运行了两次。保存文件内容如下：

{"doc_id": 1, "key_words": "Julie Wainwright, memoir, leadership, Pets.com, IPO, The RealReal, entrepreneurial, wisdom, insights.", "task_id": "cron_doc_tag_2"}
{"doc_id": 1, "key_words": "Julie Wainwright, memoir, leadership, Pets.com, setback, The RealReal, IPO, entrepreneur, wisdom", "task_id": "cron_doc_tag_1"}
{"doc_id": 2, "key_words": "DeepMind, unionize, Google, AI, workers, military, contract, employees, protests.", "task_id": "cron_doc_tag_2"}
{"doc_id": 2, "key_words": "DeepMind, unionize, Google, AI, protest, military, contract, employees, workers", "task_id": "cron_doc_tag_1"}
{"doc_id": 3, "key_words": "Google, October 25, Nest, Thermostats, updates, Europe, support, devices, homeowners.", "task_id": "cron_doc_tag_2"}
{"doc_id": 3, "key_words": "Google, Nest, thermostats, updates, support, Europe, devices, hardware, features, advancements", "task_id": "cron_doc_tag_1"}
{"doc_id": 4, "key_words": "Amazon, book sale, Independent Bookstore Day, competing, indie bookstores, ABA, market, calculated move, insensitive", "task_id": "cron_doc_tag_2"}
{"doc_id": 4, "key_words": "Amazon, book sale, Independent Bookstore Day, competition, indie bookstores, American Booksellers Association, market, calculated move, timing.", "task_id": "cron_doc_tag_1"}
{"doc_id": 5, "key_words": "Lately, app, ADHD, time management, reminders, reward system, points, users, travel plans", "task_id": "cron_doc_tag_2"}
{"doc_id": 5, "key_words": "Lately, ADHD, app, rewards, time management, points, reminders, developers, travel plans", "task_id": "cron_doc_tag_1"}

这并不符合我们的预期，我们的预期是即使这个定时任务部署在多台机器上，或者在不同地方启动时，该定时任务只需要执行一次。

使用分布式锁可以保证我们的定时任务只执行一次。

实现脚本（cron_doc_tag_1_with_lock.py）代码如下：

import os
import json
import asyncio
import redis
from dotenv import load_dotenv
from openai import AsyncOpenAI
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from apscheduler.schedulers.blocking import BlockingScheduler
from apscheduler.triggers.cron import CronTrigger

from mysql_create_table import Doc
from distribute_key_op import acquire_lock_with_timeout, release_lock

load_dotenv()

# 初始化数据库连接
engine = create_engine("mysql+pymysql://root:root@localhost:3306/orm_test")
DBSession = sessionmaker(bind=engine)

# 初始化Redis连接
redis_conn = redis.StrictRedis(host='localhost', port=6379, db=0)
LOCK_NAME = "doc_process_task"
LOCK_TIMEOUT = 600  # 锁的超时时间设置为10分钟


async def process_doc(doc_id, doc_content):
    prompt = (
        "According to the content of the document, extract less than 10 "
        "key words from the content, and return them in a string separated "
        f"by commas.\nDocument content: {doc_content}\nKey words:"
    )
    client = AsyncOpenAI(api_key=os.getenv("OPENAI_API_KEY"))
    response = await client.chat.completions.create(
        model="gpt-4o",
        messages=[{"role": "user", "content": prompt}]
    )
    result = response.choices[0].message.content
    file_path = "/Users/admin/PycharmProjects/env_test/redis_for_test/content_tags_with_lock.json"
    if not os.path.exists(file_path):
        os.system(f"touch {file_path}")
    with open(file_path, "a") as f:
        json_data = {
            "doc_id": doc_id,
            "key_words": result,
            "task_id": "cron_doc_tag_1"
        }
        f.write(json.dumps(json_data, ensure_ascii=False) + "\n")
    print(f"doc_id: {doc_id}, key_words: {result}")


async def process_all_docs():
    # 尝试获取分布式锁
    lock_identifier = acquire_lock_with_timeout(
        redis_conn,
        LOCK_NAME,
        lock_timeout=LOCK_TIMEOUT
    )

    if not lock_identifier:
        print("无法获取分布式锁，可能有其他实例正在执行任务")
        return

    try:
        # 创建新的会话
        session = DBSession()
        try:
            # 获取所有需要处理的文档
            docs = session.query(Doc).all()
            for doc in docs:
                await process_doc(doc.id, doc.content)
        finally:
            # 确保会话被关闭
            session.close()
    finally:
        # 释放分布式锁
        release_lock(redis_conn, LOCK_NAME, lock_identifier)
        print("分布式锁已释放")


def scheduled_job():
    """定时任务入口函数"""
    print("开始执行文档处理任务...")
    asyncio.run(process_all_docs())
    print("文档处理任务完成")


def main():
    """主函数：配置和启动调度器"""
    scheduler = BlockingScheduler()

    # 配置定时任务，每天21:30执行
    scheduler.add_job(
        scheduled_job,
        trigger=CronTrigger(hour=21, minute=30),
        id='process_docs_job',
        name='处理文档标签任务'
    )

    print("定时任务已启动，将在每天21:30执行...")
    try:
        scheduler.start()
    except (KeyboardInterrupt, SystemExit):
        print("定时任务已停止")


if __name__ == "__main__":
    main()

复制该脚本，重命名为cron_doc_tag_2.py，并将文件写入时"task_id"改为 "cron_doc_tag_2"。

同时运行上述两个脚本，cron_doc_tag_1.py的运行结果如下：

定时任务已启动，将在每天21:30执行...
开始执行文档处理任务...
无法获取分布式锁，可能有其他实例正在执行任务
文档处理任务完成

cron_doc_tag_2.py的运行结果如下：

定时任务已启动，将在每天21:30执行...
开始执行文档处理任务...
doc_id: 1, key_words: Julie Wainwright, memoir, leadership, Pets.com, The RealReal, entrepreneurship, IPO, setbacks, Ahara, resilience
doc_id: 2, key_words: DeepMind, unionize, Google, AI, Communication Workers Union, weapons, surveillance, Israeli military, cloud computing, protests
doc_id: 3, key_words: Google, Nest, thermostats, updates, Europe, support, generation, advancements, temperature, heating systems
doc_id: 4, key_words: Amazon, Independent Bookstore Day, American Booksellers Association, indie bookstores, sale, Bookshop.org, market, e-books, timing, Andy Hunter
doc_id: 5, key_words: Lately, ADHD, time management, rewards, points, Erik MacInnis, gamified, reminders, virtual characters, premium subscription
分布式锁已释放
文档处理任务完成

保存文件内容如下：

{"doc_id": 1, "key_words": "Julie Wainwright, memoir, leadership, Pets.com, The RealReal, entrepreneurship, IPO, setbacks, Ahara, resilience", "task_id": "cron_doc_tag_2"}
{"doc_id": 2, "key_words": "DeepMind, unionize, Google, AI, Communication Workers Union, weapons, surveillance, Israeli military, cloud computing, protests", "task_id": "cron_doc_tag_2"}
{"doc_id": 3, "key_words": "Google, Nest, thermostats, updates, Europe, support, generation, advancements, temperature, heating systems", "task_id": "cron_doc_tag_2"}
{"doc_id": 4, "key_words": "Amazon, Independent Bookstore Day, American Booksellers Association, indie bookstores, sale, Bookshop.org, market, e-books, timing, Andy Hunter", "task_id": "cron_doc_tag_2"}
{"doc_id": 5, "key_words": "Lately, ADHD, time management, rewards, points, Erik MacInnis, gamified, reminders, virtual characters, premium subscription", "task_id": "cron_doc_tag_2"}

从上面的运行结果，我们可以看出，使用分布式锁可以保证定时任务唯一执行，避免同一资源被重复消费。

短时间内避免重复执行

用户短时间内重复提交相同的请求（比如支付按钮连续点击），可以通过加锁来防止后端重复执行。

设想场景：

用户在短时间内重复点击了支付按钮，那么只有其中一个线程会执行真正的支付操作，而其它线程则不会执行支付操作。

使用分布式锁来避免短时间内重复提交，其Python实现代码如下：

import time
import redis
import threading

from distribute_key_op import acquire_lock_with_timeout, release_lock


# 连接Redis
redis_client = redis.Redis(host='localhost', port=6379, db=0)


def process_payment(user_id, order_id):
    """
    支付处理逻辑（假设耗时2秒）
    """
    print(f"[{threading.current_thread().name}] 正在处理用户 {user_id} 的订单 {order_id} 支付...")
    time.sleep(2)  # 模拟支付耗时
    print(f"[{threading.current_thread().name}] 用户 {user_id} 的订单 {order_id} 支付完成！")
    return True


def pay_order(user_id, order_id):
    try:
        lock_key = f"lock:pay:{user_id}:{order_id}"
        # 尝试加锁
        indentifier = acquire_lock_with_timeout(redis_client, lock_key, lock_timeout=120)
        if not indentifier:
            # 没拿到锁
            print(f"[{threading.current_thread().name}] 用户 {user_id} 的订单 {order_id} 正在处理中，请勿重复提交！")
            return "正在处理中，请稍后再试"

        # 成功拿到锁，处理支付
        success = process_payment(user_id, order_id)
        if success:
            return "支付成功"
        else:
            return "支付失败"

    except Exception as e:
        print(f"[{threading.current_thread().name}] 支付异常：{e}")
        return "支付异常"

    finally:
        # 释放分布式锁
        if indentifier:
            release_lock(redis_client, lock_key, indentifier)
            print(f"[{threading.current_thread().name}] 分布式锁已释放")


def simulate_multiple_payments(user_id, order_id, num_threads=5):
    """
    模拟多个线程并发支付
    """
    threads = []

    for i in range(num_threads):
        t = threading.Thread(target=pay_order, args=(user_id, order_id), name=f"Thread-{i+1}")
        threads.append(t)
        t.start()

    for t in threads:
        t.join()


if __name__ == "__main__":
    user_id = 12345
    order_id = "order_98765"

    # 模拟5个线程同时支付
    simulate_multiple_payments(user_id, order_id, num_threads=5)

输出结果如下：

[Thread-2] 正在处理用户 12345 的订单 order_98765 支付...
[Thread-4] 用户 12345 的订单 order_98765 正在处理中，请勿重复提交！
[Thread-3] 用户 12345 的订单 order_98765 正在处理中，请勿重复提交！
[Thread-5] 用户 12345 的订单 order_98765 正在处理中，请勿重复提交！
[Thread-1] 用户 12345 的订单 order_98765 正在处理中，请勿重复提交！
[Thread-2] 用户 12345 的订单 order_98765 支付完成！
[Thread-2] 分布式锁已释放

从上面的运行结果可以看出，短时间内同时提交了5次支付请求，但只有Thread-2这个线程在真正执行支付操作，其它线程因为分布式锁的原因没有拿到锁，从而没有执行支付操作。

总结

本文介绍了分布式锁的基础概念和使用场景，结合Redis基础命令，使用Python和Redis实现了分布式锁。

在分布式锁的使用场景中，笔者介绍了如何使用它来保证定时任务的唯一执行，以及如何避免在短时间内重复操作。

后续笔者介绍介绍更多Redis的使用场景，下一篇文章的主题将是如何使用Redis来实现消息队列。

欢迎关注我的公众号NLP奇幻之旅，原创技术文章第一时间推送。

欢迎关注我的知识星球“自然语言处理奇幻之旅”，笔者正在努力构建自己的技术社区。

数据库

#Redis #分布式锁

Redis进阶（一）使用Redis实现分布式锁

https://percent4.github.io/Redis进阶（一）使用Redis实现分布式锁/

作者

Jclian91

发布于

2025年4月27日

许可协议

NLP（一百二十）LiteLLM解析：构建统一大模型接口的利器下一篇