Python单实例脚本实现合理性及多实验设备管控优化问询

我需要运行一个循环A，它从目录D读取数据并执行操作，A会编辑文件并与另一对象通信，希望避免同步问题。我希望通过脚本B.py将数据加载到D中，并在A未运行时启动A，同时要能在启动A的控制台获知启动成败，遵循Python最佳实践，还想了解如何通过socket监听A。

实际场景补充：生物实验室设备管控，需管理多用户的实验，实验时长不定，设备指令不可互相覆盖，实验启停互不影响，且需跨平台兼容。

现有实现 #

A.py（循环逻辑+单实例锁） #

import time
import socket
import sys
import os

DEFAULT_PORT = 9563

try:
    # 绑定端口实现单实例锁，防止重复运行
    lock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    lock.bind(("localhost", DEFAULT_PORT))
except OSError:
    print("A is already running")
    sys.exit(3)

filename = "log.txt"
dirname = "D"
for i in range(30):
    with open(filename, "a", encoding="utf8") as f:
        num_files = len(os.listdir(dirname))
        f.write(f"idx: {i}, {time.asctime()}, {num_files} files found\n")
    time.sleep(1)

B.py（数据加载+启动A） #

import os
import sys
import platform
import time
import subprocess


dirname = "D"
# 向目录D写入数据文件
filename = str(time.time())
with open(os.path.join(dirname, filename), mode="w", encoding="utf8") as f:
    f.write("a file")

kwargs = {}
if platform.system() == "Windows":
    CREATE_NEW_PROCESS_GROUP = 0x00000200
    DETATCHED_PROCESS = 0x00000008
    kwargs.update(creationflags=DETATCHED_PROCESS | CREATE_NEW_PROCESS_GROUP)
else:
    kwargs.update(start_new_session=True)

p = subprocess.Popen([sys.executable, "A.py"],
                     stdin=subprocess.PIPE, stdout=subprocess.PIPE, **kwargs)
assert not p.poll()
print("Goodbye")
sys.exit(0)

问题咨询 #

该方案是否符合Pythonic规范？是否有更优实现方式？

现有方案的合理性与不足 #

符合Pythonic的点 #

• 用with语句管理文件资源，自动释放，符合最佳实践
• 针对Windows和类Unix系统的差异处理子进程启动参数，保证跨平台兼容性
• 用socket绑定端口实现单实例锁，思路简洁直接

存在的问题 #

1. 单实例锁隐患：socket端口可能被其他程序占用，导致A无法启动；若A异常崩溃，端口会进入TIME_WAIT状态，需等待超时才能重新绑定
2. 启动状态判断不可靠：assert not p.poll()仅能检测进程是否立即退出，无法捕获启动初期的错误（如依赖缺失、权限问题），且断言在生产环境（-O参数运行）会被禁用
3. 同步问题未解决：B写入文件时A可能读取到不完整内容；A编辑文件、与外部对象通信时无同步机制，存在冲突风险
4. Socket监听功能缺失：现有A仅用端口做锁，未实现监听和通信逻辑，无法接收外部指令或反馈状态

优化方案 #

1. 更可靠的单实例锁 #

用文件锁替代socket端口绑定，避免端口冲突问题，推荐使用跨平台库portalocker：

# A.py替换原socket锁部分
import portalocker

LOCK_FILE = "A_instance.lock"
try:
    with open(LOCK_FILE, "w") as lock_file:
        # 排他锁+非阻塞，避免等待
        portalocker.lock(lock_file, portalocker.LOCK_EX | portalocker.LOCK_NB)
except portalocker.LockException:
    print("A is already running")
    sys.exit(3)

安装依赖：pip install portalocker

2. 可靠的启动状态判断 #

B.py中移除断言，改为读取进程输出并延迟检查状态，捕获启动错误：

# B.py启动A的部分修改
p = subprocess.Popen(
    [sys.executable, "A.py"],
    stdin=subprocess.PIPE,
    stdout=subprocess.PIPE,
    stderr=subprocess.PIPE,
    text=True,
    **kwargs
)

# 延迟1秒检查进程是否存活，避免漏过启动初期错误
time.sleep(1)
if p.poll() is not None:
    stdout, stderr = p.communicate()
    print(f"A启动失败，错误信息：{stderr}")
    sys.exit(1)
print("A启动成功")

3. 解决文件同步问题 #

• 原子写入文件：B写入时先写临时文件，再重命名到目标目录（跨平台原子操作），避免A读取不完整文件：

# B.py写入文件部分修改
temp_filename = f"{filename}.tmp"
temp_path = os.path.join(dirname, temp_filename)
with open(temp_path, mode="w", encoding="utf8") as f:
    f.write("a file")
os.rename(temp_path, os.path.join(dirname, filename))

• A读取时过滤临时文件：并对共享文件操作加锁：

# A中编辑共享文件示例
import portalocker

shared_file = "shared_config.txt"
with open(shared_file, "r+", encoding="utf8") as f:
    portalocker.lock(f, portalocker.LOCK_EX)
    # 执行编辑操作
    content = f.read()
    f.seek(0)
    f.write(content.replace("old", "new"))
    f.truncate()
    portalocker.unlock(f)

4. 实现Socket监听功能 #

让A启动Socket监听，支持外部查询状态、发送停止指令：

# A.py完整优化代码
import time
import socket
import sys
import os
import portalocker

DEFAULT_PORT = 9563
LOCK_FILE = "A_instance.lock"
dirname = "D"
log_file = "log.txt"

# 单实例文件锁
try:
    with open(LOCK_FILE, "w") as lock_file:
        portalocker.lock(lock_file, portalocker.LOCK_EX | portalocker.LOCK_NB)
except portalocker.LockException:
    print("A is already running")
    sys.exit(3)

# 启动Socket监听
server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
server_socket.bind(("localhost", DEFAULT_PORT))
server_socket.listen(1)
server_socket.settimeout(1)  # 设置超时，不阻塞主循环

running = True
while running:
    # 处理Socket连接
    try:
        client_socket, addr = server_socket.accept()
        request = client_socket.recv(1024).decode("utf8").strip()
        if request == "STATUS":
            # 过滤临时文件，统计有效任务数
            num_files = len([f for f in os.listdir(dirname) if not f.endswith(".tmp")])
            response = f"Running, {num_files} pending tasks"
        elif request == "STOP":
            response = "Stopping..."
            running = False
        else:
            response = "Unknown command"
        client_socket.sendall(response.encode("utf8"))
        client_socket.close()
    except socket.timeout:
        pass  # 超时继续主循环

    # 核心业务逻辑
    with open(log_file, "a", encoding="utf8") as f:
        num_files = len([f for f in os.listdir(dirname) if not f.endswith(".tmp")])
        f.write(f"{time.asctime()}, {num_files} files found\n")
    time.sleep(1)

# 清理资源
server_socket.close()
os.remove(LOCK_FILE)

外部控制示例（查询状态）：

import socket

s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
s.connect(("localhost", DEFAULT_PORT))
s.sendall(b"STATUS")
print(s.recv(1024).decode("utf8"))
s.close()

5. 实验室场景适配建议 #

• 多用户隔离：为每个用户创建独立子目录，或用任务队列（如celery）管理实验任务，自动调度设备资源
• 设备指令互斥：对设备通信模块加全局锁，确保同一时间仅一条指令发送给设备
• 日志与监控：用标准库logging替代手动写日志，支持分级记录；添加进程守护（如supervisord、systemd），确保A崩溃后自动重启

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