使用文件锁flock实现进程互斥

最近写了一个python脚本,对Android设备进行自动化测试。当重复执行脚本时,如果不对访问设备进行锁控制,不同的脚本可能会同时向设备发出指令,导致测试失败。所以尝试用flock实现了简单的脚本互斥。当获取锁失败时,flocked函数会raise OsError。此时说明已经有脚本在运行。

import fcntl
from contextlib import contextmanager
import logging

logger = logging.getLogger(__name__)

@contextmanager
def flocked(lock_file):
    with open(lock_file, 'w') as fd:
        try:
            # Lock file. Raise OsError if failed to lock.
            fcntl.flock(fd, fcntl.LOCK_EX | fcntl.LOCK_NB)
            logger.info('Acquired lock for %s', lock_file)
            yield
        finally:
            fcntl.flock(fd, fcntl.LOCK_UN)
            logger.info('Released lock for %s', lock_file)

try:
    with flocked('/tmp/lock_file'):
        # Do something.
except OsError:
       logger.exception('Failed to lock file')
最后编辑于
©著作权归作者所有,转载或内容合作请联系作者
平台声明:文章内容(如有图片或视频亦包括在内)由作者上传并发布,文章内容仅代表作者本人观点,简书系信息发布平台,仅提供信息存储服务。

推荐阅读更多精彩内容