当前位置：首页 > article >正文

使用 Python 实现分布式任务锁：详解与示例

article 2024/11/1 0:58:13

在这里插入图片描述
💝💝💝欢迎莅临我的博客，很高兴能够在这里和您见面！希望您在这里可以感受到一份轻松愉快的氛围，不仅可以获得有趣的内容和知识，也可以畅所欲言、分享您的想法和见解。

推荐：「stormsha的主页」👈，「stormsha的知识库」👈持续学习，不断总结，共同进步，为了踏实，做好当下事儿~
专栏导航
- Python系列: Python面试题合集，剑指大厂
- Git系列: Git操作技巧
- GO系列: 记录博主学习GO语言的笔记，该笔记专栏尽量写的试用所有入门GO语言的初学者
- 数据库系列: 详细总结了常用数据库 mysql 技术点,以及工作中遇到的 mysql 问题等
- 运维系列: 总结好用的命令，高效开发
- 算法与数据结构系列: 总结数据结构和算法,不同类型针对性训练,提升编程思维
非常期待和您一起在这个小小的网络世界里共同探索、学习和成长。💝💝💝 ✨✨ 欢迎订阅本专栏 ✨✨

💖The Start💖点点关注，收藏不迷路💖
📒文章目录
- - 一、什么是任务锁？
  - 二、实现分布式任务锁的步骤
  - - 1. 引入必要的模块
    - 2. 定义任务锁的上下文管理器
    - 3. 使用任务锁
    - 4. 日志记录
  - 三、改进建议
  - - 1. 锁争抢等待
    - 2. 可配置的 Redis 连接
    - 3. 锁定时间动态配置
  - 四、总结
  - 参考

在分布式系统中，任务锁（Task Lock）是一种常见的技术手段，用于防止多个进程或线程同时执行相同的任务。通过任务锁，可以确保任务的唯一性和一致性。在这篇博文中，我们将介绍如何使用 Python 和 Redis 实现分布式任务锁，并提供详细的代码示例和改进建议。

一、什么是任务锁？

任务锁是一种机制，用于确保在同一时间内只有一个进程或线程可以执行特定的任务。任务锁通常使用分布式存储系统（如 Redis）来实现，因为分布式存储系统能够提供快速、高效的锁操作。在任务锁的帮助下，我们可以避免任务的重复执行，从而保证数据的一致性和操作的原子性。

二、实现分布式任务锁的步骤

在本节中，我们将详细介绍如何使用 Redis 和 Python 实现一个分布式任务锁。

1. 引入必要的模块

首先，我们需要引入 contextlib 中的 contextmanager 模块，以及用于处理异常的 traceback 模块。

from contextlib import contextmanager
import traceback
import logging

2. 定义任务锁的上下文管理器

接下来，我们定义一个名为 task_lock 的上下文管理器，用于在任务执行前后管理锁的设置和释放。

@contextmanager
def task_lock(name, redis_conn, expire=60):
    """
    锁定任务以防止重复执行
    :param name: 任务名，用于区分不同的任务
    :param redis_conn: Redis 连接对象
    :param expire: 锁的过期时间，默认60秒
    :return:

    用法示例：
    >>> with task_lock('test', redis_conn) as e:
    >>>    if e:
    >>>        print('doing something')
    >>>    else:
    >>>        print('task locked, skip')
    """
    name = 'task_lock:' + name
    try:
        is_locked = redis_conn.set(name, 1, nx=True, ex=expire)
        if is_locked:
            logging.info(f"Task {name} locked successfully.")
        else:
            logging.info(f"Task {name} is already locked.")
    except Exception as err:
        redis_conn.delete(name)
        logging.error(f"Failed to lock task {name}: {err}")
        raise Exception(f"任务加锁失败 {err} {traceback.format_exc()}")

    try:
        yield is_locked
    finally:
        redis_conn.delete(name)
        logging.info(f"Task {name} unlocked.")

3. 使用任务锁

下面是一个使用任务锁的示例。在这个示例中，我们尝试使用 task_lock 上下文管理器锁定任务，如果锁定成功，则执行任务；否则，跳过任务执行。

import redis
redis_conn = redis.StrictRedis(host='localhost', port=6379, db=0)

with task_lock('test_task', redis_conn) as locked:
    if locked:
        print('Task is running...')
        # 执行任务的代码
    else:
        print('Task is already locked, skipping...')