Python中Semaphore锁机制应用求助：relay_CAN共享锁定实现

Hey，我看你在搞relay_CAN共享的项目，遇到了跨设备锁定继电器的问题，先点破核心误区：你用的threading.Semaphore、Lock()、Condition()都是单进程内的线程同步工具，它们的锁状态只存在当前Python进程的内存里，完全没法在不同终端/设备之间共享——这就是你的代码无效的根本原因！

先说说你现有代码里的明显bug：

sem = threading.Semaphore()
# ...
if sem.acquire(False):
    sem.acquire()  # 这里直接会阻塞死！
    print "locking.."
    try:
        #todo
    finally:
        print"locked successefully"

acquire(False)是非阻塞式获取信号量，成功后你又调用sem.acquire()（默认阻塞），这时候信号量的计数已经被减到0了，当前线程会一直卡在这里，永远等不到释放，逻辑完全错了。

正确的跨设备同步思路 #

要实现跨设备的继电器互斥锁定，你得用所有设备都能访问的共享锁存储，给你推荐几个实用方案：

1. 共享文件锁（适合简单场景，依赖共享存储） #

如果你的设备能访问同一个共享文件系统（比如NAS、挂载的网络盘），可以用操作系统的文件锁来实现：

import fcntl
import time

# 所有设备都能访问的锁文件路径
RELAY_LOCK_PATH = "/shared_storage/relay_can_lock.lock"

def acquire_relay_lock(blocking=True):
    lock_file = open(RELAY_LOCK_PATH, "w")
    try:
        if blocking:
            # 阻塞等待，直到拿到锁
            fcntl.flock(lock_file, fcntl.LOCK_EX)
        else:
            try:
                # 非阻塞，拿不到直接返回失败
                fcntl.flock(lock_file, fcntl.LOCK_EX | fcntl.LOCK_NB)
            except BlockingIOError:
                return False, lock_file
        return True, lock_file
    except Exception as e:
        print(f"获取锁失败: {str(e)}")
        lock_file.close()
        return False, None

def release_relay_lock(lock_file):
    try:
        fcntl.flock(lock_file, fcntl.LOCK_UN)
        lock_file.close()
    except Exception as e:
        print(f"释放锁失败: {str(e)}")

def main():
    import argparse
    parser = argparse.ArgumentParser()
    parser.add_argument("--connect", help="锁定继电器并连接设备", action="store_true")
    parser.add_argument("--disconnect", help="解锁继电器并断开设备", action="store_true")
    args = parser.parse_args()

    if args.connect:
        acquired, lock_file = acquire_relay_lock(blocking=True)
        if acquired:
            print("✅ 继电器已锁定，开始使用设备...")
            # 这里替换成你的继电器连接、CAN通信逻辑
            try:
                # 模拟设备使用过程，实际业务中替换为你的代码
                while True:
                    time.sleep(1)
            except KeyboardInterrupt:
                print("\n收到中断信号，准备解锁...")
            finally:
                release_relay_lock(lock_file)
                print("✅ 继电器已解锁")
        else:
            print("❌ 继电器已被其他设备锁定，无法获取")
    elif args.disconnect:
        print("🔓 尝试解锁继电器...")
        # 严谨点的话要验证当前设备是否持有锁，这里简化处理
        _, lock_file = acquire_relay_lock(blocking=False)
        if lock_file:
            release_relay_lock(lock_file)
            print("✅ 解锁成功")
        else:
            print("❌ 继电器未被锁定或当前设备无权限解锁")

if __name__ == "__main__":
    main()

• 优点：零额外依赖，实现简单
• 缺点：必须依赖稳定的共享文件系统，锁的可靠性受文件系统影响

2. Redis分布式锁（推荐，适合复杂多设备场景） #

如果你的设备都能访问同一个Redis服务器，这是最可靠的方案，还能避免死锁：
先安装Redis库：pip install redis

import redis
import time
from redis.lock import Lock

# 配置你的Redis服务器信息
REDIS_HOST = "192.168.1.100"  # 替换成你的Redis地址
REDIS_PORT = 6379
REDIS_DB = 0
RELAY_LOCK_KEY = "relay_can_lock"

def get_redis_client():
    return redis.Redis(host=REDIS_HOST, port=REDIS_PORT, db=REDIS_DB)

def main():
    import argparse
    parser = argparse.ArgumentParser()
    parser.add_argument("--connect", help="锁定继电器并连接设备", action="store_true")
    parser.add_argument("--disconnect", help="解锁继电器并断开设备", action="store_true")
    args = parser.parse_args()

    redis_client = get_redis_client()
    # 设置锁超时时间5分钟，防止设备崩溃导致死锁
    lock = Lock(redis_client, RELAY_LOCK_KEY, timeout=300)

    if args.connect:
        if lock.acquire(blocking=True):
            print("✅ 继电器已锁定，开始使用设备...")
            # 替换为你的继电器业务逻辑
            try:
                while True:
                    time.sleep(1)
                    # 如果设备使用时间超过超时时间，定期刷新锁有效期
                    lock.extend(300)
            except KeyboardInterrupt:
                print("\n收到中断信号，准备解锁...")
            finally:
                lock.release()
                print("✅ 继电器已解锁")
        else:
            print("❌ 继电器已被其他设备锁定，请等待解锁")
    elif args.disconnect:
        print("🔓 尝试解锁继电器...")
        # 验证是否是当前设备持有锁（Redis锁会自动记录持有者）
        if lock.release():
            print("✅ 解锁成功")
        else:
            print("❌ 继电器未被锁定或当前设备无权限解锁")

if __name__ == "__main__":
    main()

• 优点：可靠性高，支持跨网络设备，自带超时自动释放机制，避免死锁
• 缺点：需要部署Redis服务器

3. MQTT消息队列（适合物联网设备场景） #

如果你的设备是物联网终端，可以用MQTT来实现锁的同步和通知：

• 所有设备订阅relay/can/lock/status主题，获取锁的实时状态
• 设备尝试锁定时，向relay/can/lock/request主题发送锁定请求
• 可以用一个中心服务来维护锁状态，或者利用MQTT的保留消息来存储锁的当前状态
• 解锁时发布解锁消息到主题，其他等待中的设备收到通知后可以尝试获取锁

重要提示 #

• 防止死锁：不管用哪种方案，一定要设置锁的超时时间，避免设备意外崩溃后锁一直无法释放
• 权限验证：解锁时要确认当前设备是锁的持有者，别让无关设备误解锁
• 状态一致性：锁的状态必须存在所有设备都能访问的共享介质里，绝对不能用进程内的同步原语！

内容的提问来源于stack exchange，提问作者mathais