You need to enable JavaScript to run this app.
优惠活动
大模型
产品
解决方案
定价
更多

Python独立脚本部署临界资源避免竞态条件的问题排查

问题根本原因

multiprocessing.Lock 仅能在同一父进程派生出来的关联子进程之间实现互斥,本质是内核维护的、通过进程继承关系传递的匿名同步对象。
你通过两个终端并行启动脚本时,相当于启动了两个完全独立、无继承关系的Python主进程,每个进程启动时都会自行初始化一个全新的、独立的Lock实例,两个锁互不可见、完全不关联,自然无法拦截对方进入临界区。
你当前的测试逻辑中,锁是在每个脚本的if __name__ == '__main__'代码块中单独创建的,仅对当前脚本进程派生的子进程生效,跨独立进程完全没有互斥能力。

落地方案

针对树莓派多SSH用户操作GPIO的场景,优先选实现简单、可靠性高的跨独立进程互斥方案:

方案1:系统文件锁(零额外依赖,最快落地)

Linux系统自带的文件锁是内核级维护的同步对象,所有进程只要争抢同一个固定路径的文件锁,就能实现全局互斥,Python标准库fcntl原生支持该能力,不需要安装任何第三方包。
首先封装全局GPIO锁:

import fcntl
# 固定锁文件路径,所有操作GPIO的进程都用这个路径
GPIO_LOCK_PATH = "/var/lock/gpio_exclusive.lock"

class GpioExclusiveLock:
    def __enter__(self):
        self._lock_file = open(GPIO_LOCK_PATH, "w")
        # 加排他锁,锁被占用时自动阻塞等待
        fcntl.flock(self._lock_file, fcntl.LOCK_EX)
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        # 操作完成释放锁,关闭文件句柄
        fcntl.flock(self._lock_file, fcntl.LOCK_UN)
        self._lock_file.close()

把所有GPIO操作的临界区代码用这个锁包裹即可,替换你原来的multiprocessing.Lock:

def main1(input2, input1, num, val):
    with GpioExclusiveLock():
        print(input2)
        print(input1)
        time.sleep(int(input1))
        # 此处放置实际GPIO读写操作
        val.value = val.value + 1

使用时注意提前给锁文件路径配置普通用户的读写权限,避免普通SSH用户因权限不足无法拿锁。如果需要自定义等待超时,可以搭配fcntl.LOCK_NB非阻塞加锁+循环重试实现。

方案2:独立GPIO常驻服务(稳定性最优,适合长期运行)

如果后续要扩展权限控制、操作审计、故障自动恢复能力,建议单独写一个常驻后台的系统服务,由该服务唯一持有GPIO的操作权限,所有SSH用户的操作请求通过本地Unix套接字、命名管道等方式发送给服务,服务端维护请求队列,同一时间仅处理一个GPIO操作。
这种方案彻底避免了多进程直接操作硬件的状态冲突问题,所有GPIO操作逻辑集中维护,排查问题也更方便。

避坑提醒

不要使用multiprocessing.Manager实现的跨进程锁,该方案需要先启动独立的Manager服务进程,你仍然需要解决不同独立启动的用户进程如何连接到同一个Manager服务的问题,实现复杂度远高于文件锁,没有必要。

内容的提问来源于stack exchange,提问作者LearnerJ

相关产品推荐
方舟 Agent Plan

超全模态模型 × Harness 升级,最新支持 Deepseek-V4.1-Flash、GLM-5.3 系列、Doubao-Seedream-5.0-pro、Kimi-K3 (部分), 限时 9.9 元起

最近更新时间:2026.09.03 06:24:31