Перейти к содержанию

Создаем первое приложение 💻#

Поздравляем, вы настроили FastStream и RabbitMQ! 🎉 Теперь пора создать первое приложение, которое будет принимать текстовые сообщения из очереди и обрабатывать их. В этом разделе мы напишем простой код, разберем, как он работает, и настроим обработку сообщений. А еще мы объясним, как правильно использовать декораторы broker.subscriber. Готовы начать кодить? Погнали! 🚀

Что мы сделаем? 📋#

Мы создадим приложение, которое:

  • Слушает очередь input-queue в RabbitMQ 📥.
  • Принимает текстовое сообщение, преобразует его в верхний регистр 🔠.
  • Логирует результат обработки.

Это базовый пример, который покажет, как FastStream упрощает работу с брокерами сообщений. Публикация сообщений будет рассмотрена в разделе Взаимодействие между сервисами.

Шаг 1: Написание кода ✍️#

Создайте файл app.py в папке faststream-tutorial и добавьте следующий код:

app.py
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
from faststream import FastStream, Logger
from faststream.rabbit import RabbitBroker

# Создаем брокер RabbitMQ
broker = RabbitBroker("amqp://guest:guest@localhost:5672/")
# Создаем приложение FastStream
app = FastStream(broker)


# Подписываемся на очередь "input-queue"
@broker.subscriber("input-queue")
async def handle_message(msg: str, logger: Logger) -> None:
    logger.info(f"Получено сообщение: {msg}")
    logger.info(f"Обработано: {msg.upper()}")

Что здесь происходит? 🤔

  • RabbitBroker: Объект для подключения к RabbitMQ. Мы указываем адрес amqp://guest:guest@localhost:5672/ (логин, пароль и порт по умолчанию).
  • FastStream: Основное приложение, которое управляет брокером.
  • @broker.subscriber("input-queue"): Декоратор, реализованный как метод брокера, который регистрирует функцию handle_message как подписчика на очередь input-queue. Он указывает, что функция будет обрабатывать входящие сообщения.
  • handle_message: Функция-обработчик, которая принимает строку, логирует её и выводит обработанную версию (в верхнем регистре).

Шаг 2: Запуск приложения ▶️#

Убедитесь, что RabbitMQ запущен (см. Установка и настройка). Если вы используете Docker, проверьте, что контейнер работает:

docker ps

Запустите приложение FastStream:

faststream run app:app

Вы увидите вывод:

INFO     - FastStream app starting...
INFO     - input-queue  |            - `handle_message` waiting for messages
INFO     - FastStream app started successfully! To exit press CTRL+C

Это означает, что приложение слушает очередь input-queue и готово обрабатывать сообщения! 🎉

Шаг 3: Практическое задание 📚#

Попробуйте закрепить знания:

  1. Измените функцию handle_message, чтобы она логировала другой префикс, например, Ответ: вместо Обработано:.
  2. Добавьте в handle_message логирование длины входного сообщения (len(msg)) и проверьте результат.
  3. Убедитесь, что RabbitMQ запущен, и попробуйте отправить сообщение в input-queue через веб-интерфейс RabbitMQ (http://localhost:15672). Проверьте логи приложения.

Что дальше? 🗺️#

Вы создали и запустили своё первое приложение с FastStream! 🎉 Теперь вы знаете, как использовать декоратор broker.subscriber для обработки сообщений. В следующем разделе мы узнаем, как настроить взаимодействие между сервисами, чтобы публиковать и обрабатывать результаты в других очередях. Перейдите к Взаимодействие между сервисами, чтобы освоить этот процесс.

Если у вас есть идеи, вопросы или нужна помощь, загляните в официальную документацию FastStream, пишите в Telegram или Discord. Продолжайте творить! 🚀