FastAPI에서 다중 처리를 수행하는 방법

async def 끝점

ProcessPoolExecutor 와 함께 loop.run_in_executor 를 사용 하여 별도의 프로세스에서 함수를 시작할 수 있습니다 .

@app.post("/async-endpoint")
async def test_endpoint():
    loop = asyncio.get_event_loop()
    with concurrent.futures.ProcessPoolExecutor() as pool:
        result = await loop.run_in_executor(pool, cpu_bound_func)  # wait result

def 끝점

def엔드 포인트는 별도의 스레드에서 암시 적 으로 실행 되므로 모듈 다중 처리 및 동시 처리 의 모든 기능을 사용할 수 있습니다 . 내부 def기능 await은 사용할 수 없습니다. 견본:

@app.post("/def-endpoint")
def test_endpoint():
    ...
    with multiprocessing.Pool(3) as p:
        result = p.map(f, [1, 2, 3])

@app.post("/def-endpoint/")
def test_endpoint():
    ...
    with concurrent.futures.ProcessPoolExecutor(max_workers=3) as executor:
      results = executor.map(f, [1, 2, 3])

참고 : 엔드 포인트에서 프로세스 풀을 생성하고 많은 수의 스레드를 생성하면 요청 수가 증가함에 따라 응답 속도가 느려질 수 있다는 점을 기억해야합니다.

즉석에서 실행

별도의 프로세스에서 함수를 실행하고 즉시 결과를 기다리는 가장 쉽고 기본적인 방법은 ProcessPoolExecutor 와 함께 loop.run_in_executor 를 사용하는 것 입니다 .

아래 예에서와 같이 풀은 응용 프로그램이 시작될 때 생성 될 수 있으며 응용 프로그램 종료시 종료하는 것을 잊지 마십시오. 풀에서 사용되는 프로세스 수는 max_workers ProcessPoolExecutor 생성자 매개 변수를 사용하여 설정할 수 있습니다 . 하면 max_workers된다 None또는 제공하지, 그것은 기계의 프로세서 수에 기본 설정됩니다.

이 접근 방식의 단점은 요청 처리기 (경로 작업)가 별도의 프로세스에서 계산이 완료 될 때까지 대기하는 반면 클라이언트 연결은 열려 있다는 것입니다. 그리고 어떤 이유로 연결이 끊어지면 결과를 반환 할 곳이 없습니다.

import asyncio
from concurrent.futures.process import ProcessPoolExecutor
from fastapi import FastAPI

from calc import cpu_bound_func

app = FastAPI()


async def run_in_process(fn, *args):
    loop = asyncio.get_event_loop()
    return await loop.run_in_executor(app.state.executor, fn, *args)  # wait and return result


@app.get("/{param}")
async def handler(param: int):
    res = await run_in_process(cpu_bound_func, param)
    return {"result": res}


@app.on_event("startup")
async def on_startup():
    app.state.executor = ProcessPoolExecutor()


@app.on_event("shutdown")
async def on_shutdown():
    app.state.executor.shutdown()

배경으로 이동

일반적으로 CPU 바운드 작업은 백그라운드에서 실행됩니다. FastAPI는 실행 할 수있는 기능 제공 백그라운드 작업이 실행되도록 한 후 시작하고 비동기 CPU 바인딩 작업의 결과를 기다릴 수있는 응답, 내부를 반환.

예를 들어이 경우 "Accepted"(HTTP 코드 202) 의 응답 과 고유 한 작업을 즉시 반환 ID하고 백그라운드에서 계산을 계속할 수 있으며 클라이언트는 나중에 this를 사용하여 작업의 상태를 요청할 수 있습니다 ID.

BackgroundTasks특히 몇 가지 기능을 제공합니다 (종속성 포함). 그리고 그들에서 당신은 모든 작업이 완료되었을 때만 정리되는 종속성에서 얻은 리소스를 사용할 수 있으며 예외의 경우 올바르게 처리 할 수 ​​있습니다. 이것은이 다이어그램 에서 더 명확하게 볼 수 있습니다 .

다음은 최소한의 작업 추적을 수행하는 예입니다. 실행중인 응용 프로그램의 한 인스턴스가 가정됩니다.

import asyncio
from concurrent.futures.process import ProcessPoolExecutor
from http import HTTPStatus

from fastapi import BackgroundTasks
from typing import Dict
from uuid import UUID, uuid4
from fastapi import FastAPI
from pydantic import BaseModel, Field

from calc import cpu_bound_func


class Job(BaseModel):
    uid: UUID = Field(default_factory=uuid4)
    status: str = "in_progress"
    result: int = None


app = FastAPI()
jobs: Dict[UUID, Job] = {}


async def run_in_process(fn, *args):
    loop = asyncio.get_event_loop()
    return await loop.run_in_executor(app.state.executor, fn, *args)  # wait and return result


async def start_cpu_bound_task(uid: UUID, param: int) -> None:
    jobs[uid].result = await run_in_process(cpu_bound_func, param)
    jobs[uid].status = "complete"


@app.post("/new_cpu_bound_task/{param}", status_code=HTTPStatus.ACCEPTED)
async def task_handler(param: int, background_tasks: BackgroundTasks):
    new_task = Job()
    jobs[new_task.uid] = new_task
    background_tasks.add_task(start_cpu_bound_task, new_task.uid, param)
    return new_task


@app.get("/status/{uid}")
async def status_handler(uid: UUID):
    return jobs[uid]


@app.on_event("startup")
async def startup_event():
    app.state.executor = ProcessPoolExecutor()


@app.on_event("shutdown")
async def on_shutdown():
    app.state.executor.shutdown()

더 강력한 솔루션

위의 모든 예는 아주 간단했다,하지만 당신은 분산 컴퓨팅 무거운에 대한 좀 더 강력한 시스템을 필요로하는 경우에, 당신은 메시지 브로커 따로 볼 수 RabbitMQ, Kafka, NATS셀러리 같은 themthem을 사용하는 등 및 라이브러리를.