如何限制Linux系统中指定进程user_proc的总运行实例数上限

核心思路 #

使用系统级的全局流量控制器限制user_proc的实例数，所有调用进程共用同一个计数器，获取到配额才启动进程，进程结束后自动归还配额，排队进程会在配额释放后自动触发启动。

• 信号量初始值设置为你需要的max_user_proc_count（比如示例中的5）
• 所有用户的Python调用逻辑共用同一个全局信号量/锁池
• 无论进程正常退出还是异常崩溃，都保证配额自动归还，不会出现资源泄漏

方案1：POSIX命名信号量实现（推荐） #

性能高、逻辑简洁，需要安装第三方依赖posix-ipc

pip install posix-ipc

代码示例 #

import posix_ipc
import subprocess

# 配置参数
MAX_PROC_COUNT = 5
# 全局信号量名称，所有调用进程共用同一个
SEMAPHORE_NAME = "/user_proc_limit_sem"

def run_user_proc(*args):
    # 创建/获取全局命名信号量，所有用户都有访问权限
    sem = posix_ipc.Semaphore(
        SEMAPHORE_NAME,
        flags=posix_ipc.O_CREAT,
        mode=0o666,
        initial_value=MAX_PROC_COUNT
    )
    try:
        # 阻塞等待获取配额
        sem.acquire()
        # 启动user_proc进程，等待运行结束
        proc = subprocess.Popen(["user_proc"] + list(args))
        proc.wait()
        return proc.returncode
    finally:
        # 无论进程运行结果如何，强制归还配额
        sem.release()

if __name__ == "__main__":
    # 此处传入user_proc需要的参数即可
    run_user_proc("arg1", "arg2")

注意事项 #

• 若需要调整阈值，修改MAX_PROC_COUNT后，执行rm /dev/shm/sem.user_proc_limit_sem删除旧信号量即可，下次运行会自动创建新的
• 权限0o666可根据场景调整为对应用户组权限，满足安全要求

方案2：文件锁实现（无第三方依赖） #

不需要安装额外库，兼容性更强，适合不能随便装依赖的生产环境

代码示例 #

import fcntl
import subprocess
import os
import time

# 配置参数
MAX_PROC_COUNT = 5
LOCK_DIR = "/var/run/user_proc_locks"
# 初始化锁目录，所有用户都有读写权限
os.makedirs(LOCK_DIR, mode=0o777, exist_ok=True)

def run_user_proc(*args):
    lock_fd = None
    # 轮询获取空闲锁
    while True:
        for i in range(MAX_PROC_COUNT):
            lock_path = os.path.join(LOCK_DIR, f"proc_{i}.lock")
            try:
                fd = os.open(lock_path, os.O_CREAT | os.O_RDWR, 0o666)
                # 非阻塞加锁，成功则获取到配额
                fcntl.flock(fd, fcntl.LOCK_EX | fcntl.LOCK_NB)
                lock_fd = fd
                break
            except BlockingIOError:
                os.close(fd)
                continue
        if lock_fd:
            break
        # 没有空闲配额，等待1秒再重试
        time.sleep(1)
    
    try:
        proc = subprocess.Popen(["user_proc"] + list(args))
        proc.wait()
        return proc.returncode
    finally:
        # 释放锁，归还配额
        fcntl.flock(lock_fd, fcntl.LOCK_UN)
        os.close(lock_fd)

if __name__ == "__main__":
    run_user_proc("arg1", "arg2")

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