多进程开发项目：如何判断目标进程是否运行？获取CWD或用标记

阿华AIGC实验室

2026-5-19

当然可行，咱们一步步拆解你的问题，先聊聊通过进程CWD验证的方案，再重点说说用锁文件或端口做标记的靠谱做法。

一、通过进程CWD确认目标进程是否可行？

完全可行，但有几个需要注意的细节：

实现方式：不同操作系统获取进程CWD的方法不同：
- Linux/Unix：可以读取/proc/<pid>/cwd符号链接，或者用lsof -p <pid> | grep cwd命令直接获取
- Windows：通过WMI查询Win32_Process类的WorkingSetDirectory属性，或者用tasklist结合工具解析
局限性：
- 权限问题：如果脚本没有足够权限访问目标进程的CWD（比如进程是其他用户运行的），可能获取失败
- 动态变更：如果目标进程运行过程中主动切换了工作目录，那通过CWD判断就会失效
适用场景：如果你的目标进程启动后不会修改工作目录，且脚本有足够权限访问，这种方法是有效的，但单独用它做唯一判断不够稳妥，建议结合后面的标记方案一起用。

二、用端口或锁文件做进程运行标记的具体方案

这两种都是工业界常用的进程存活标记方案，各有优劣，你可以根据自己的场景选择：

方案1：锁文件（推荐本地进程场景）

锁文件的核心思路是：进程启动时创建一个独占的锁文件，退出时自动删除；其他进程通过检查锁文件的存在性+关联PID的存活状态，来判断目标进程是否在运行。

实现步骤：

选择一个固定路径作为锁文件位置（比如/var/run/your_app.lock或者用户目录下的.your_app.lock）
进程启动时，尝试创建锁文件：
- 用独占锁打开文件（避免多个进程同时创建），写入当前进程的PID
- 如果创建失败（文件已存在），检查文件中的PID对应的进程是否还在运行：
  - 如果PID存活，说明目标进程已在运行
  - 如果PID不存在，说明是残留的锁文件，删除后重新创建
进程正常退出时删除锁文件；如果异常崩溃，可以在下次启动时清理残留文件

代码示例（Python）：

import os
import sys
import fcntl

LOCK_FILE = "/tmp/my_app.lock"

def acquire_lock():
    try:
        lock_fd = os.open(LOCK_FILE, os.O_CREAT | os.O_WRONLY | os.O_EXCL)
        os.write(lock_fd, f"{os.getpid()}".encode())
        os.close(lock_fd)
        return True
    except FileExistsError:
        # 检查锁文件中的PID是否存活
        with open(LOCK_FILE, 'r') as f:
            pid = int(f.read().strip())
        try:
            os.kill(pid, 0)  # 发送空信号检查进程是否存在
            return False  # 进程仍在运行
        except OSError:
            # PID不存在，删除残留锁文件并重试
            os.unlink(LOCK_FILE)
            return acquire_lock()

if not acquire_lock():
    print("目标进程已在运行")
    sys.exit(1)

# 后续进程逻辑
try:
    while True:
        # 你的业务代码
        pass
finally:
    os.unlink(LOCK_FILE)

优缺点：

✅ 优点：不需要网络资源，实现简单，适合本地单进程守护场景
❌ 缺点：异常崩溃可能残留锁文件，需要额外的PID检查逻辑；多用户场景下要注意文件权限

方案2：端口绑定（推荐跨进程/轻量检查场景）

端口绑定的思路是：目标进程启动时绑定一个固定的本地端口，其他进程通过尝试绑定该端口（或连接该端口）来判断进程是否在运行——如果绑定失败（Address already in use），说明目标进程已在运行。

实现步骤：

选择一个不常用的本地端口（比如10086，或者在配置文件中指定）
目标进程启动时，创建一个socket并绑定该端口，设置SO_REUSEADDR（可选，避免进程异常退出后端口处于TIME_WAIT状态导致无法绑定）
其他进程尝试绑定同一个端口，如果失败则判定目标进程已运行

代码示例（Python）：

import socket
import sys

CHECK_PORT = 10086

def is_process_running():
    sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
    try:
        sock.bind(('127.0.0.1', CHECK_PORT))
        sock.close()
        return False
    except socket.error as e:
        if e.errno in (48, 10048):  # Linux/Windows的端口占用错误码
            return True
        else:
            raise

if is_process_running():
    print("目标进程已在运行")
    sys.exit(1)

# 绑定端口保持进程运行
server_sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
server_sock.bind(('127.0.0.1', CHECK_PORT))
server_sock.listen(1)

try:
    while True:
        # 你的业务代码
        pass
finally:
    server_sock.close()