如何通过Python脚本执行PM2启停状态命令并过滤指定服务输出？

好问题！我来帮你梳理几种可行的方案，从最直接的到更便捷的都有：

方案1：用subprocess捕获并解析PM2的JSON输出 #

你之前用的subprocess.call()只能拿到命令的退出码，没法获取输出内容——得换成subprocess.run()或者subprocess.check_output()才能捕获标准输出。而且PM2本身支持输出JSON格式的数据，这比解析表格文本靠谱多了，不会因为输出格式变动导致脚本失效。

示例代码 #

import subprocess
import json

def get_pm2_service_status(service_name):
    try:
        # 执行pm2命令，指定--json参数输出结构化数据
        result = subprocess.run(
            ['pm2', 'status', service_name, '--json'],
            capture_output=True,
            text=True,
            check=True
        )
        # 解析JSON字符串为Python字典
        status_data = json.loads(result.stdout)
        # pm2 --json返回的是数组，单个服务取第一个元素即可
        if status_data:
            return status_data[0]
        return None
    except subprocess.CalledProcessError as e:
        print(f"执行命令出错: {e.stderr}")
        return None

# 使用示例
service_info = get_pm2_service_status('someService')
if service_info:
    print(f"服务名称: {service_info['name']}")
    print(f"当前状态: {service_info['status']}")
    print(f"进程ID: {service_info['pid']}")
    print(f"CPU使用率: {service_info['cpu']}%")
    print(f"内存占用: {service_info['memory']/1024/1024:.2f} MB")

方案2：使用专门的Python PM2库 #

如果你不想自己处理命令调用，确实有第三方Python库可以直接和PM2交互，比如pm2-api。不过要注意，这类库的维护可能跟不上PM2的版本更新，兼容性需要自己测试。

步骤示例 #

1. 先安装库：

pip install pm2-api

2. 调用示例代码：

from pm2_api import PM2

# 初始化PM2客户端
pm2_client = PM2()

# 获取所有服务的状态
all_services = pm2_client.status()

# 筛选出目标服务
target_service = next((s for s in all_services if s['name'] == 'someService'), None)

if target_service:
    print(f"服务{target_service['name']}的状态是{target_service['status']}")
    print(f"重启次数: {target_service['restart_time']}")

方案3：调用PM2的HTTP API（或通过curl） #

PM2自带了HTTP API服务，你可以先通过pm2 web命令启动它（默认监听9615端口），之后就能通过HTTP请求获取状态数据了——不管是用Python的requests库直接调用，还是通过curl命令再解析都可以。

用requests直接调用API #

import requests

def get_pm2_status_via_api(service_name):
    try:
        # 请求PM2的API接口
        response = requests.get('http://localhost:9615/status')
        response.raise_for_status()  # 检查请求是否成功
        status_data = response.json()
        
        # 从返回的进程列表中筛选目标服务
        target_service = next(
            (s for s in status_data['processes'] if s['name'] == service_name),
            None
        )
        return target_service
    except requests.exceptions.RequestException as e:
        print(f"调用API出错: {e}")
        return None

# 使用示例
service_info = get_pm2_status_via_api('someService')
if service_info:
    print(f"服务状态: {service_info['pm2_env']['status']}")
    print(f"当前运行路径: {service_info['pm2_env']['pm_cwd']}")

用curl命令配合subprocess调用 #

如果你偏好使用curl，也可以通过subprocess执行curl命令后解析输出：

import subprocess
import json

def get_pm2_status_via_curl(service_name):
    try:
        result = subprocess.run(
            ['curl', '-s', 'http://localhost:9615/status'],
            capture_output=True,
            text=True,
            check=True
        )
        status_data = json.loads(result.stdout)
        target_service = next(
            (s for s in status_data['processes'] if s['name'] == service_name),
            None
        )
        return target_service
    except subprocess.CalledProcessError as e:
        print(f"执行curl出错: {e.stderr}")
        return None

总结 #

综合来看，方案1是最推荐的：不需要额外安装依赖，兼容性最好，而且JSON格式的数据处理起来非常灵活。如果你的场景需要频繁和PM2交互，方案2可以考虑；如果需要远程获取状态，方案3更合适。

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