I/O-bound — операция, скорость выполнения которой ограничена скоростью подсистемы ввода-вывода. К таким задачам можно отнести выполнение запросов по сети, операции с базой, чтение/запись файла на диск. 

Например, HTTP-запрос:

r = requests.get('https://api.github.com/user', auth=('user', 'pass'))

CPU-bound — операция, скорость выполнения которой ограничена скоростью CPU. В web-сервисах это обычно работа с медиа: обрезка, ресайз, конвертация картинок.

from contextlib import asynccontextmanager
from typing import Annotated
from fastapi import Depends


class BaseAccessor:
   def __init__(self, store: "Store") -> None:
       self.store = store
       self.config = settings

   async def connect(self) -> None:
       pass

   async def disconnect(self) -> None:
       pass


class Store:
   def __init__(self) -> None:
       # чтобы исключить циклический импорт
       from app.core.accessor import CoreAccessor
       from app.redis.accessor import RedisAccessor
       from app.worker.accessor import WorkerAccessor

       self.core = CoreAccessor(self)
       self.redis = RedisAccessor(self)
       self.worker = WorkerAccessor(self)

   async def connect(self) -> None:
       await self.redis.connect()
       await self.worker.connect()
       await self.core.connect()

   async def disconnect(self) -> None:
       await self.core.disconnect()
       await self.worker.disconnect()
       await self.redis.disconnect()


_store: Store | None = None


def get_store() -> Store:
   assert _store, "Store is not initialized"
   return _store


async def connect_to_store() -> Store:
   global _store

   if not _store:
       _store = Store()
       await _store.connect()

   return _store


async def disconnect_from_store() -> None:
   global _store

   if _store:
       await _store.disconnect()
       _store = None


@asynccontextmanager
async def store_lifespan() -> AsyncGenerator[Store, None]:
   await connect_to_store()
   try:
       yield get_store()
   finally:
       await disconnect_from_store()


StoreDep = Annotated[Store, Depends(get_store)]

import arq
import arq.connections


# фоновая задача: функция для отправки эл. почты в фоне
async def send_email(ctx: dict, email: str, message: str = "Hello") -> None:
   store: Store = ctx["store"]
   …


# класс для доступа к фоновым задачам
class WorkerAccessor(AbstractTaskAccessor):
   def __init__(self, *args: Any, **kwargs: Any) -> None:
       super().__init__(*args, **kwargs)

	  # инициализируем пул задач arq
       self.job_pool: arq.ArqRedis | None = None

   async def connect(self) -> None:
	  # устанавливаем соединение с Redis
       self.job_pool = await arq.create_pool(
           arq.connections.RedisSettings(
			# настраиваем параметры соединения
               host=self.config.REDIS_DSN.host,
               port=self.config.REDIS_DSN.port,
               username=self.config.REDIS_DSN.username,
               password=self.config.REDIS_DSN.password,
               database=self.config.REDIS_DSN.path.removeprefix("/"),
           )
       )

   async def disconnect(self) -> None:
       """Для закрытия соединения"""
       if self.job_pool:
           await self.job_pool.close()

   async def run(self) -> None:
       """Для запуска воркера в другом сервисе"""
      
       async def on_startup(ctx: dict) -> None:
		 # устанавливаем объект хранилища в контексте
           ctx["store"] = self.store

       worker = arq.Worker(
		 # указываем фоновые задачи
           functions=[send_email],
           on_startup=on_startup,
           max_jobs=self.config.ARQ_CONCURRENCY,
           health_check_interval=60,
           handle_signals=False,
           redis_pool=self.job_pool,
       )
       try:
		 # запускаем воркер
           await worker.main()
       finally:
		 # закрываем воркер
           await worker.close()
          
   async def send_email(self, email: str, message: str = "Hello") -> None:
       # ставим задачу в очередь на исполнение
       await self.job_pool.enqueue_job(send_email.__name__, email, message=message)

        
# запускаем фоновую задачу отправки эл. почты, вариант 1
@router.post("")
async def hello(store: StoreDep) -> JSONResponse:
   task = BackgroundTask(store.worker.arq.send_email, "support@mail.ru", message="Hello")
   return JSONResponse({"status": "ok"}, background=task)


# запускаем фоновую задачу отправки эл. почты, вариант 2
@router.post("")
async def hello(store: StoreDep) -> dict:
   await store.worker.arq.send_email("support@mail.ru", message="Hello")
   return {"status": "ok"}

async def main() -> None:
   async with store_lifespan() as store:
       await store.worker.arq.run()


if __name__ == "__main__":
   asyncio.run(main())

import saq
from aiohttp.web_runner import AppRunner, TCPSite
from redis.asyncio import Redis
from saq.web.aiohttp import create_app


# фоновая задача
async def send_email(ctx: dict, email: str, message: str = "Hello") -> None:
   store: Store = ctx["store"]
   …


class WorkerkAccessor(BaseAccessor):
   def __init__(self, *args: Any, **kwargs: Any) -> None:
       super().__init__(*args, **kwargs)

       self.queue: saq.Queue | None = None

   async def connect(self) -> None:
       client = Redis.from_url(self.config.REDIS_DSN.__str__())
       self.queue = saq.Queue(client)

   async def disconnect(self) -> None:
       if self.queue:
           await self.queue.disconnect()

   async def run(self, web: bool = False, port: int = 8090) -> None:
      """Для запуска воркера в другом сервисе"""
      
       async def on_startup(ctx: dict) -> None:
           ctx["store"] = self.store

       worker = saq.Worker(
           self.queue,
           functions=[send_email],  # указываем фоновые задачи
           startup=on_startup,
           concurrency=self.config.SAQ_CONCURRENCY,
       )

       if web:
           # запускаем дашборд
           app = create_app([self.queue])
           task = asyncio.create_task(worker.start())

           try:
               runner = AppRunner(app)
               await runner.setup()
               site = TCPSite(runner, port=port)
               await site.start()

               await asyncio.Event().wait()
           finally:
               await worker.stop()
               task.cancel()
       else:
           await worker.start()

   async def send_email(self, email: str, message: str = "Hello") -> None:
       # ставим задачу в очередь на исполнение
       await self.queue.enqueue(send_email.__name__, email=email, message=message)


@router.post("")
async def hello(store: StoreDep) -> JSONResponse:
   task = BackgroundTask(store.worker.saq.send_email, "support@mail.ru", message="Hello")
   return JSONResponse({"status": "ok"}, background=task)


# или так
@router.post("")
async def hello(store: StoreDep) -> dict:
   await store.worker.saq.send_email("support@mail.ru", message="Hello")
   return {"status": "ok"}

async def main() -> None:
   async with store_lifespan() as store:
       await store.worker.saq.run(web=True)


if __name__ == "__main__":
   asyncio.run(main())

import faststream
from faststream.redis import RedisBroker


class FastStream(faststream.FastStream):
   def __init__(self, *args: Any, **kwargs: Any) -> None:
       super().__init__(*args, **kwargs)

       self.on_startup(self.connect)
       self.on_shutdown(self.disconnect)

   async def connect(self, *_: Any, **__: Any) -> None:
       await connect_to_store()

       await self.broker.connect(url=settings.REDIS_DSN.__str__())

   async def disconnect(self, *_: Any, **__: Any) -> None:
       await disconnect_from_store()

       await self.broker.close()


broker = RedisBroker()

# для запуска воркера через терминал
faststream_app = FastStream(broker)

# название фоновой задачи
SEND_EMAIL_TOPIC = "send_email"


# фоновая задача
@broker.subscriber(SEND_EMAIL_TOPIC)
async def send_email(email: str, message: str) -> None:
   …


class WorkerAccessor(BaseAccessor):
   async def connect(self) -> None:
       await broker.connect(url=self.config.REDIS_DSN.__str__())

   async def disconnect(self) -> None:
       await broker.close()

   async def send_email(self, email: str, message: str = "Hello") -> None:
       # ставим задачу в очередь на исполнение
       await broker.publish(
           message={"email": email, "message": message},
           channel=SEND_EMAIL_TOPIC,
       )




@router.post("")
async def hello(store: StoreDep) -> JSONResponse:
   task = BackgroundTask(store.worker.faststream.send_email, "support@mail.ru", message="Hello")
   return JSONResponse({"status": "ok"}, background=task)


# или так
@router.post("")
async def hello(store: StoreDep) -> dict:
   await store.worker.faststream.send_email("support@mail.ru", message="Hello")
   return {"status": "ok"}

faststream run app.worker.faststream:faststream_app –workers=1

faststream docs serve app.worker.faststream:faststream_app

import celery
from celery import signals
from functools import wraps


class Celery(celery.Celery):
   def __init__(self, *args: Any, **kwargs: Any) -> None:
       super().__init__(*args, **kwargs)
      
       # тут хранятся функции для удобства тестирования (eager execution)
       self.functions: dict[str, Callable[..., Any]] = {}
      
       # цикл событий, чтобы из синхронной функции вызывать асинхронную
       self.loop = asyncio.get_event_loop()

   def connect(self, *_: Any, **__: Any) -> None:
       self.loop.run_until_complete(connect_to_store())

   def disconnect(self, *_: Any, **__: Any) -> None:
       self.loop.run_until_complete(disconnect_from_store())

   def task(
       self,
       task: Callable[..., Awaitable[T]] | None = None,
       **opts: Any,
   ) -> Callable:
       # декоратор от celery
       create_task = super().task

       def decorator(func: Callable[..., Awaitable[T]]) -> Callable[..., T]:
           @create_task(**opts)  # регистрируем задачу
           @wraps(func)
           def wrapper(*args: Any, loop: AbstractEventLoop | None = None, **kwargs: Any) -> T:
               # для случаев, когда очень хочется выполнить задачу сразу (без apply_async)
               loop = loop or self.loop

               # выполняем асинхронную функцию в цикле событий
               return loop.run_until_complete(func(*args, **kwargs))
          
           # запоминаем функцию для удобства тестирования
           self.functions[wrapper.name] = func

           return wrapper

       if task:
           return decorator(task)

       return decorator


celery_app = Celery("celery", broker=settings.REDIS_DSN.__str__())
celery_app.autodiscover_tasks(packages=["app.worker.async_celery.tasks"])

# подключаем store
signals.worker_process_init.connect(celery_app.connect)
signals.worker_process_shutdown.connect(celery_app.disconnect)

# фоновая задача
@celery_app.task
async def send_email(email: str, message: str) -> None:
   …


@router.post("")
async def hello() -> JSONResponse:
   task = BackgroundTask(send_email.delay, "support@mail.ru", message="Hello")
   return JSONResponse({"status": "ok"}, background=task)


# варианты:
@router.post("")
async def hello() -> dict:
   # примеры постановки задачи в очередь на исполнение
   send_email.delay("support@mail.ru", message="Hello")
   send_email.apply_async(("support@mail.ru",), {"message": "Hello"}, countdown=5)
   send_email.s("support@mail.ru", message="Hello").apply_async(countdown=5)

   # пример запуска задачи в текущем процессе
   send_email("support@mail.ru", message="Hello", loop=asyncio.get_running_loop())

   return {"status": "ok"}

from contextlib import ExitStack
import nest_asyncio
import pytest


# чтобы не ловить ошибки о том, что цикл событий уже запущен
nest_asyncio.apply()


@pytest.fixture(scope="session", autouse=True)
def celery_eager_execution(store: Store) -> Generator[None, None, None]:
   # регистрируем все фоновые задачи
   from app.worker.async_celery import tasks  # noqa: F401

   def execute_task(name: str) -> Callable[..., Any]:
       # достаем функцию для этой задачи и выполняем ее в качестве side_effect у мока
       func = celery_app.functions[name]

       def wrapper(args: Any, kwargs: Any, **_: Any) -> Any:
           loop = asyncio.get_running_loop()
           return loop.run_until_complete(func(*args, **kwargs))

       return wrapper
  
   # создаем моки для каждой задачи
   patches = [
       patch.object(task, "apply_async", side_effect=execute_task(name))
       for name, task in celery_app.tasks.items()
   ]
  
   # одновременно применяем все моки
   with ExitStack() as stack:
       for p in patches:
           stack.enter_context(p)
       yield

celery -A app.worker.async_celery.app:celery_app worker –concurrency=3

celery -A app.worker.async_celery.app:celery_app flower

from fastapi import FastAPI
from prometheus_fastapi_instrumentator import Instrumentator

from app.api.router import router


app = FastAPI(title="FastAPI async tasks")
app.include_router(router)

# все метрики будут доступны по пути /metrics
Instrumentator().instrument(app).expose(app)

from prometheus_async.aio import count_exceptions, time
from prometheus_client import Counter, Histogram


EXECUTION_TIME = Histogram(
   "task_execution_seconds",
   "Task execution time",
   labelnames=["task_name"],  # чтобы в метриках видеть название фоновой задачи
)
TASKS_FAILED = Counter(
   "tasks_failed",
   "Tasks failed",
   labelnames=["task_name"],  # чтобы в метриках видеть название фоновой задачи
)


# декоратор для удобного сбора метрик
def export_task_metrics(func: Callable) -> Callable:
   task_name = func.__name__
  
   @count_exceptions(TASKS_FAILED.labels(task_name=task_name))  # подсчет количества ошибок
   @time(EXECUTION_TIME.labels(task_name=task_name))  # подсчет времени выполнения
   @wraps(func)
   def wrapper(*args: Any, **kwargs: Any) -> Any:
       return func(*args, **kwargs)

   return wrapper

@export_task_metrics
async def send_email(email: str, message: str) -> None:
   ...

from prometheus_async.aio.web import start_http_server_in_thread


async def main() -> None:
   start_http_server_in_thread(port=settings.ARQ_EXPORTER_PORT)

   async with store_lifespan() as store:
       await store.worker.arq.run()


if __name__ == "__main__":
   asyncio.run(main())

# prometheus.yml

global:
 scrape_interval: 15s


scrape_configs:
 - job_name: "fastapi"
   static_configs:
     - targets: ["fastapi:8000"]


 - job_name: "arq"
   static_configs:
     - targets: ["arq:8001"]


 - job_name: "saq"
   static_configs:
     - targets: ["saq:8002"]


 - job_name: "faststream"
   static_configs:
     - targets: ["faststream:8003"]


 - job_name: "async-celery"
   static_configs:
     - targets: ["async-celery-flower:5555"]

docker build -t fastapi-async-tasks .

# docker-compose.yml

version: "3.8"


x-tasks-common: &tasks-common
   image: fastapi-async-tasks
   volumes:
       - .:/app  # для разработки
   depends_on:
       - fastapi
   restart: on-failure
   networks:
       - default
   deploy:
       resources:
           limits:
               memory: 200M
               cpus: "0.2"


services:
   redis:
       image: redis:7.2.4
       ports:
           - "6379:6379"
       restart: on-failure
       networks:
           - default


   prometheus:
       image: prom/prometheus
       ports:
           - "9090:9090"
       volumes:
           - .prometheus/prometheus.yml:/etc/prometheus/prometheus.yml
       command:
           -  --config.file=/etc/prometheus/prometheus.yml
       restart: on-failure
       networks:
           - default


   grafana:
       image: grafana/grafana
       ports:
           - "3000:3000"
       volumes:
           - type: volume
             source: grafana
             target: /var/lib/grafana
       environment:
           - GF_SECURITY_ADMIN_USER=admin
           - GF_SECURITY_ADMIN_PASSWORD=admin
           - GF_INSTALL_PLUGINS=redis-app
       user: "0"
       restart: on-failure
       networks:
           - default


   fastapi:
       image: fastapi-async-tasks
       ports:
           - "8000:8000"
       volumes:
           - .:/app
       depends_on:
           - redis
           - prometheus
           - grafana
       restart: on-failure
       networks:
           - default
       deploy:
           resources:
               limits:
                   memory: 200M
                   cpus: "0.2"
       command:
           - uvicorn
           - app.main:app
           - --host=0.0.0.0
           - --port=8000
           - --reload
           - --workers=${FASTAPI_WORKERS}


   arq:
       <<: *tasks-common
       ports:
           - "8001:8001"  # metrics
       command:
           - python
           - -m
           - worker_arq


   saq:
       <<: *tasks-common
       ports:
           - "8082:8082"  # web dashboard
           - "8002:8002"  # metrics
       command:
           - python
           - -m
           - worker_saq


   faststream:
       <<: *tasks-common
       ports:
           - "8003:8003"  # metrics
       command:
           - faststream
           - run
           - app.worker.faststream.app:faststream_app
           - --workers=${FASTSTREAM_CONCURRENCY}


   async-celery:
       <<: *tasks-common
       command:
           - celery
           - -A
           - app.worker.async_celery.app:async_celery_app
           - worker
           - --loglevel=info
           - --concurrency=${ASYNC_CELERY_CONCURRENCY}


   async-celery-flower:
       image: fastapi-async-tasks
       ports:
           - "5555:5555"
       volumes:
           - .:/app
       depends_on:
           - async-celery
       restart: on-failure
       networks:
           - default
       command:
           - celery
           - -A
           - app.worker.async_celery.app:async_celery_app
           - flower


volumes:
   grafana:


networks:
   default:
       driver: bridge

docker-compose up -d

# locust/master.conf

headless = true  # управление через консоль
expect-workers = 10  # сначала дожидаемся старта 10 процессов
host = http://localhost:8000  # адрес FastAPI

# locust/background.py

from random import shuffle
from locust import between, HttpUser, task


# для тестирования Background Task
class TaskUser(HttpUser):
   wait_time = between(0.001, 0.001)

   @task
   def execute(self):
       endpoint = os.environ.get("endpoint")
       endpoints = (
           [endpoint]
           if endpoint
           else ["io/simple", "io/sync", "io/thread", "cpu/simple", "cpu/process"]
       )
       shuffle(endpoints)

       for endpoint in endpoints:
           self.client.post(f"/api/v1/incr/{endpoint}/background", json={"value": 1})

# locust/benchmark.py

from random import shuffle
from locust import between, HttpUser, task


# для тестирования остальных библиотек
class TaskUser(HttpUser):
   wait_time = between(0.001, 0.001)

   @task
   def execute(self):
       endpoint = os.environ.get("endpoint")
       endpoints = (
           [endpoint]
           if endpoint
           else ["io/simple", "io/sync", "io/thread", "cpu/simple", "cpu/process"]
       )

       worker = os.environ.get("worker")
       workers = (
           [worker]
           if worker
           else ["arq", "saq", "faststream", "async-celery"]
       )

       params = [(endpoint, worker) for endpoint in endpoints for worker in workers]
       shuffle(params)

       for endpoint, worker in params:
           self.client.post(f"/api/v1/incr/{endpoint}/{worker}", json={"value": 1})

env endpoint=io/simple locust --config locust/master.conf -f locust/background.py -i 10000 -u 100