如何搭建可与持续运行子进程交互的Python REST服务

核心需求 #

• 基于Python实现REST服务端
• 存在一个持续运行不终止的后台任务，支持通过stdin接收指令、stdout输出结果
• 用户可通过REST API向后台任务发送请求并获取返回结果，示例场景为查询每秒自动+1的计数值

选型说明 #

• 无需使用Celery等额外任务调度组件，降低学习成本
• 采用FastAPI作为web框架，自带原生异步支持，搭配asyncio内置subprocess能力实现和后台任务的非阻塞交互，无额外依赖
• 全程仅需2个脚本即可跑通最小可用流程

代码实现 #

1. 常驻后台计数任务脚本（counter.py） #

该脚本独立运行，每秒自动将计数+1，收到stdin输入的get\n指令时，会向stdout输出当前计数值

import sys
import time
import select

def main():
    count = 0
    # 关闭stdout缓冲，避免输出滞留在缓冲区无法读取
    sys.stdout.reconfigure(line_buffering=True)
    while True:
        count += 1
        time.sleep(1)
        # 非阻塞检查是否有输入指令
        ready, _, _ = select.select([sys.stdin], [], [], 0)
        if ready:
            cmd = sys.stdin.readline().strip()
            if cmd == 'get':
                print(f"current_count:{count}", flush=True)

if __name__ == "__main__":
    main()

2. REST服务端脚本（main.py） #

启动时自动拉起常驻后台计数任务，暴露GET接口供用户查询计数

import asyncio
from fastapi import FastAPI

app = FastAPI()
# 全局保存后台任务的subprocess实例
background_proc = None

# 服务启动钩子：拉起后台计数任务
@app.on_event("startup")
async def startup_event():
    global background_proc
    background_proc = await asyncio.create_subprocess_exec(
        "python", "counter.py",
        stdin=asyncio.subprocess.PIPE,
        stdout=asyncio.subprocess.PIPE,
        stderr=asyncio.subprocess.DEVNULL
    )

# 获取计数接口
@app.get("/get-count")
async def get_count():
    # 向后台任务stdin发送查询指令
    background_proc.stdin.write(b"get\n")
    await background_proc.stdin.drain()
    # 读取后台任务返回的结果
    result = await background_proc.stdout.readline()
    return {"result": result.decode().strip()}

# 服务关闭钩子：销毁后台任务
@app.on_event("shutdown")
async def shutdown_event():
    if background_proc:
        background_proc.kill()
        await background_proc.wait()

运行与测试步骤 #

1. 安装依赖：pip install fastapi uvicorn
2. 将两个脚本放在同一目录下，执行命令启动服务：uvicorn main:app --host 0.0.0.0 --port 8000
3. 调用接口测试，访问http://localhost:8000/get-count即可获取当前最新计数值，多次调用可看到数值持续增长

关键说明 #

• 后台任务在服务启动时就拉起，全程不会终止，除非服务关闭
• 用asyncio的subprocess实现非阻塞交互，请求不会出现超时问题，接口响应速度快
• 如果要改用Flask实现，只需要把异步逻辑换成同步subprocess，加上线程锁避免多请求同时读写stdin/stdout出现冲突即可，核心交互逻辑完全一致

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