Отчёт по лабораторной работе №7: Упаковка приложения в Docker, работа с источниками данных и очереди задач

1. Цель работы

Целью работы является контейнеризация распределенной системы на базе FastAPI, разделение монолита на микросервисы, организация межсервисного взаимодействия по протоколу HTTP и внедрение асинхронных очередей задач с использованием Taskiq и Redis. Также ставилась задача реализации периодических (cron) задач для автоматизации сбора данных.

2. Стек технологий и структура проекта

В дополнение к стеку первой лабораторной работы были внедрены: - aiohttp — для высокопроизводительного асинхронного парсинга. - Taskiq — современный асинхронный менеджер задач (альтернатива Celery). - Redis 7 — брокер сообщений и хранилище результатов. - httpx — для межсервисного взаимодействия внутри Docker-сети. - BeautifulSoup4 — для извлечения данных из HTML.

Новая структура распределенной системы: - App (Port 8102): Основной шлюз и API. - Parser Service (Port 8104): Изолированный микросервис только для парсинга. - Worker: Обработчик фоновых задач. - Scheduler: Планировщик периодических задач. - Redis: Брокер. - PostgreSQL: Единое хранилище.

3. Архитектура и контейнеризация (Task 1)

В ходе работы проект был разделен на независимые сервисы. Для микросервиса парсера был написан отдельный Dockerfile на базе python:3.11-slim. Взаимодействие между контейнерами настроено через внутреннюю сеть Docker.

Фрагмент docker-compose.yml (новые компоненты):

services:
  # Микросервис парсера (aiohttp)
  parser:
    build: ./parser_app
    container_name: parser
    ports:
      - "8104:8000"

  # Брокер сообщений
  redis:
    image: redis:7-alpine
    container_name: redis

  # Обработчик задач (Taskiq)
  worker:
    build: .
    command: taskiq worker app.tkq:broker app.tasks --fs-discover
    depends_on: [redis, db, parser]

  # Планировщик cron-задач
  scheduler:
    build: .
    command: taskiq scheduler app.tkq:scheduler
    depends_on: [redis, worker]

4. Реализация микросервиса и прямой вызов (Task 2)

Согласно заданию, логика парсинга вынесена в отдельное приложение. Основное приложение обращается к нему по HTTP. Реализована обработка ошибок: если целевой сайт недоступен, микросервис возвращает корректный статус ошибки, а шлюз пробрасывает его клиенту.

Пример прямого вызова парсера (app/api/parser.py):

@router.post("/sync")
async def parse_sync(url: str):
    async with httpx.AsyncClient() as client:
        # Обращение к соседнему контейнеру по имени сервиса
        response = await client.get(f"http://parser:8000/parse?url={url}")
        if response.status_code != 200:
            raise HTTPException(status_code=response.status_code)
        return response.json()

Результат успешного вызова парсера через Swagger зафиксирован на скриншоте ниже.

Результат синхронного парсинга в Swagger

5. Очереди задач и периодические задания (Task 3)

Для выполнения длительных операций реализована фоновая обработка. Использование Taskiq позволило сохранить полную асинхронность на всех этапах (FastAPI -> Redis -> Worker -> PostgreSQL).

Фоновые задачи: Клиент получает task_id мгновенно, в то время как воркер скачивает данные и сохраняет их в БД.
Миграции: Добавлена новая модель ParsedPage для хранения результатов. Процесс миграции автоматизирован через Alembic.
Периодические задачи: Настроено расписание (Cron) для автоматического парсинга сайта qutoq.site каждую минуту.

Реализация периодической задачи (app/tasks.py):

@broker.task(schedule=[{"cron": "* * * * *"}])
async def periodic_parse_qutoq(
    session: Annotated[AsyncSession, TaskiqDepends(get_async_session)]
):
    # Автоматический вызов парсинга по расписанию
    await run_parsing_task(url="http://qutoq.site", session=session)

Ниже представлена визуализация работы планировщика и результат накопления данных в базе данных (каждую минуту добавляется новая запись).

Результаты в БД и работа планировщика

6. Выводы

В ходе лабораторной работы была реализована микросервисная архитектура: - Приложение разделено на функциональные блоки (API и Parser), упакованные в Docker. - Налажена межсервисная связь внутри Docker-сети. - Внедрена асинхронная очередь задач на базе Taskiq и Redis, что позволило разгрузить основной поток приложения. - Реализован механизм периодического сбора данных (Scheduler), работающий в автономном режиме. - Обеспечена сохранность данных в PostgreSQL через асинхронное взаимодействие воркера и БД.