如何在Python中实现多节点多进程并发删除文件行并执行命令？

这个场景我之前在集群运维的时候碰过不少，先给你划个重点：操作系统绝对不会自动帮你处理跨节点的文件并发修改，本地进程的同步机制在多节点环境里完全不管用，必须自己做跨节点的同步控制，或者换个更靠谱的架构。

先说说你的原始思路的问题：直接让不同节点的进程去读写同一个共享文件，很容易出现多个进程同时修改文件，导致内容错乱、命令重复执行或者丢失的情况，必须加跨节点有效的排他锁。

下面给你两个可行的方案，你可以根据自己的集群环境选：

方案一：用跨节点文件锁实现共享文件的任务分发 #

如果你一定要用文件来存命令，那必须用跨节点生效的文件锁，核心思路是：同一时间只允许一个进程（不管哪个节点的）修改命令文件，其他进程都等着拿锁。

这里要注意：你的共享存储（比如NFS）必须支持跨节点的文件锁——比如NFS v4是原生支持的，NFS v3需要额外配置lockd和statd服务才能用。

给你写个Python的示例代码，逻辑很简单：

1. 每个worker进程先尝试获取共享锁文件的排他锁
2. 拿到锁之后，才去读取命令文件，取出第一行，把剩下的内容写回去
3. 释放锁之后再执行命令（别拿着锁执行命令，不然其他进程都要等你命令执行完，完全失去并发意义）

import fcntl
import os
import subprocess
import multiprocessing

def worker_task():
    command_path = "/mnt/shared/commands.txt"
    # 锁文件必须放在同一个共享存储上
    lock_path = "/mnt/shared/command_lock.lock"

    while True:
        current_cmd = None
        # 用with块自动管理锁的释放：进程崩溃或正常退出都会自动释放锁
        with open(lock_path, 'w') as lock_file:
            try:
                # 加排他锁，拿不到就阻塞等待
                fcntl.flock(lock_file, fcntl.LOCK_EX)

                # 打开命令文件处理内容
                with open(command_path, 'r+') as cmd_file:
                    lines = cmd_file.readlines()
                    if not lines:
                        # 没有命令了，退出循环
                        break
                    # 取第一行命令，记得去掉换行符
                    current_cmd = lines[0].strip()
                    # 把剩下的行写回文件，然后截断多余内容
                    cmd_file.seek(0)
                    cmd_file.writelines(lines[1:])
                    cmd_file.truncate()
            except Exception as e:
                print(f"进程{os.getpid()}处理文件出错: {str(e)}")
                continue

        # 拿到命令后释放了锁，现在可以安全执行命令了
        if current_cmd:
            print(f"进程{os.getpid()}开始执行命令: {current_cmd}")
            try:
                # 这里用subprocess.run执行命令，你可以根据需求加stdout/stderr的捕获
                subprocess.run(current_cmd, shell=True, check=True)
            except subprocess.CalledProcessError as e:
                print(f"进程{os.getpid()}执行命令失败: {current_cmd}，错误信息: {str(e)}")
                # 如果需要重试，可以在这里把命令重新加回文件（记得再加锁）
            except Exception as e:
                print(f"进程{os.getpid()}执行命令时发生未知错误: {str(e)}")

if __name__ == "__main__":
    # 每个节点启动和CPU核心数一样多的worker进程
    worker_count = os.cpu_count() or 4
    processes = []
    for _ in range(worker_count):
        p = multiprocessing.Process(target=worker_task)
        p.start()
        processes.append(p)
    
    # 等待所有进程结束
    for p in processes:
        p.join()

方案一的注意事项 #

• 必须确保共享文件系统支持跨节点锁，否则锁只会在本地节点生效，跨节点还是会乱
• 别拿着锁执行命令！锁只应该在修改文件的时候持有，执行命令的时间要放出去，不然并发效率会极低
• 如果命令执行失败，你可以在捕获异常后重新把命令写入文件（记得加锁），避免任务丢失

方案二：换用集中式任务队列（更推荐） #

其实用共享文件做任务分发是比较原始的方案，容易出问题还效率低。我更推荐你用集中式任务队列来替代，比如用Redis的阻塞队列：

1. 先把所有命令一次性导入Redis的列表里
2. 每个节点的worker进程用blpop命令从队列里拿任务（这个命令是原子性的，Redis单线程天然保证跨节点的并发安全）
3. 拿到任务后直接执行，完全不用管锁的问题

这种方案比文件靠谱太多：没有锁的问题，不会出现文件损坏，任务不会丢，效率也高很多。Python里用redis库就能轻松实现，代码比文件锁的逻辑还简单。

最后再提醒你一句：如果一定要用文件方案，一定要做充分的测试，比如模拟某个节点进程崩溃、网络短暂中断的情况，看看会不会出现命令丢失或重复执行的问题。

内容来源于stack exchange