Python自动化平台开发中怎么解决高并发任务冲突

初学者常尝试使用 Python 标准库中的 threading.Lock 或 asyncio.Lock 来避免冲突。但在现代自动化平台中，应用通常采用 Gunicorn 多工作进程、Celery 分布式节点或多 Docker 容器运行，彼此拥有独立的内存空间与解释器进程。进程内锁无法跨越进程边界，必须借助中心化协调媒介（如 Redis 或数据库）进行跨进程互斥。

import time
import uuid
import threading
import logging
import redis

logger = logging.getLogger("TaskLockManager")

class RedisResourceLock:
    """
    高并发自动化平台专用的分布式资源锁
    支持自动看门狗续期与 Lua 安全原子释放
    """
    RELEASE_LUA = """
    if redis.call("get", KEYS[1]) == ARGV[1] then
        return redis.call("del", KEYS[1])
    else
        return 0
    end
    """

    def __init__(self, redis_client: redis.Redis, resource_name: str, 
                 lease_ms: int = 10000, heartbeat_interval_s: float = 3.0):
        self.client = redis_client
        self.key = f"lock:resource:{resource_name}"
        self.lease_ms = lease_ms
        self.heartbeat_interval = heartbeat_interval_s
        self.lock_token = str(uuid.uuid4())
        self._renew_thread = None
        self._stop_renew = threading.Event()
        self._release_script = self.client.register_script(self.RELEASE_LUA)

    def acquire(self, timeout_s: float = 5.0) -> bool:
        start_time = time.time()
        while time.time() - start_time < timeout_s:
            # 使用 NX PX 指令实现互斥加锁与超时兜底
            acquired = self.client.set(self.key, self.lock_token, nx=True, px=self.lease_ms)
            if acquired:
                self._start_watchdog()
                logger.info(f"成功获取资源锁 [{self.key}], token: {self.lock_token}")
                return True
            time.sleep(0.1)  # 适当休眠后重试
        return False

    def _start_watchdog(self):
        """后台守护线程：定期续租以防长任务被意外释放"""
        self._stop_renew.clear()
        def watchdog_loop():
            while not self._stop_renew.wait(self.heartbeat_interval):
                # 只有当前锁 token 匹配时才续约
                check_script = """
                if redis.call("get", KEYS[1]) == ARGV[1] then
                    return redis.call("pexpire", KEYS[1], ARGV[2])
                else
                    return 0
                end
                """
                res = self.client.eval(check_script, 1, self.key, self.lock_token, self.lease_ms)
                if not res:
                    logger.warning(f"续约失败，锁已被抢占或过期: {self.key}")
                    break

        self._renew_thread = threading.Thread(target=watchdog_loop, daemon=True)
        self._renew_thread.start()

    def release(self):
        """终止看门狗并执行原子解锁"""
        self._stop_renew.set()
        if self._renew_thread and self._renew_thread.is_alive():
            self._renew_thread.join(timeout=1.0)
        
        # 必须使用 Lua 脚本比对 token，确保不误删别人持有的锁
        result = self._release_script(keys=[self.key], args=[self.lock_token])
        logger.info(f"释放锁 [{self.key}] 结果: {result}")
        return bool(result)

    def __enter__(self):
        if not self.acquire():
            raise RuntimeError(f"无法获取资源锁 [{self.key}], 存在并发冲突")
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        self.release()

from distributed_lock import RedisResourceLock
import redis

r_client = redis.Redis(host="127.0.0.1", port=6379, db=0)

def run_device_automation_task(task_id: str, device_id: str):
    logger.info(f"收到执行指令，准备锁定设备: {device_id}")
    
    try:
        # 尝试对特定设备资源加锁，限制竞争粒度
        with RedisResourceLock(r_client, resource_name=device_id, lease_ms=8000):
            logger.info(f"成功独占设备 {device_id}，开始执行自动化脚本...")
            # 模拟执行测试脚本逻辑
            execute_test_suite(task_id, device_id)
            logger.info(f"自动化用例执行完毕，退出代码块自动释放锁")
    except RuntimeError as err:
        logger.warning(f"任务 [{task_id}] 争抢设备 {device_id} 失败，延后重试或标记排队中: {err}")
        raise  # 触发队列的退避重试 (Exponential Backoff)

-- 仅当数据库中的当前版本与内存读取到的预期版本一致时，才允许更新
UPDATE platform_tasks 
SET 
    status = 'SUCCESS', 
    result_summary = '用例全量通过',
    version = version + 1, 
    updated_at = NOW()
WHERE 
    id = 'task_20261101_0089' 
    AND version = 2;

避免将锁键设置为如 lock:platform:auto_test 这样的全局大锁。锁的粒度越细，系统并发性能越高。应将锁细化至“最小不可拆分物理资源”，例如 lock:device:serial_9876 或 lock:user_session:1042，确保不同设备上的用例能够全并行推进。

抢锁失败时，绝不要使用紧凑的 while True: pass 空转循环，这会迅速打满 CPU 并压垮 Redis。应加入随机抖动退避（如 time.sleep(0.05 + random.uniform(0, 0.05))），并在达到最大重试次数后主动放弃并转入延迟队列。

若执行 Worker 被宿主机 OOM-Killer 杀死或网络断连，Watchdog 线程自然终止。Redis 上的键将在租约超时（lease_ms）后自动过期。平台需具备死信检测机制，自动识别那些长时间处于 RUNNING 却已失去 Redis 活跃标记的任务，重置其状态或告警通知。

在高并发运行期间，必须对“任务抢锁耗时”、“加锁失败次数”以及“重试等待时长”打点监控。当某个资源的锁争抢等待时长持续上升时，说明该硬件资源已成为测试流水线的瓶颈，应及时扩充硬件节点或动态调整调度权重。