Обмен сообщениями Pub-Sub (публикация-подписка) – это популярный шаблон связи, используемый в распределенных системах, где издатели отправляют сообщения центральному брокеру, а подписчики получают эти сообщения от брокера. FastAPI, современная веб-инфраструктура Python, обеспечивает отличную поддержку для создания высокопроизводительных API. В этой статье мы углубимся в различные методы реализации обмена сообщениями Pub-Sub с использованием FastAPI, а также приведем примеры кода.

1. Использование Redis Pub-Sub:
   Redis — это хранилище структур данных в памяти, которое поддерживает обмен сообщениями Pub-Sub. Чтобы использовать Redis Pub-Sub с FastAPI, вы можете использовать библиотеку aioredis. Вот пример простого приложения FastAPI, которое демонстрирует, как публиковать сообщения и подписываться на них с помощью Redis:

import asyncio
from fastapi import FastAPI
import aioredis
app = FastAPI()
redis = await aioredis.create_redis_pool("redis://localhost")
@app.post("/publish/{channel}")
async def publish_message(channel: str, message: str):
    await redis.publish(channel, message)
    return {"status": "Message published"}
@app.websocket("/subscribe/{channel}")
async def subscribe_to_channel(channel: str, websocket: WebSocket):
    pubsub = await redis.subscribe(channel)
    channel = pubsub[0]
    while await channel.wait_message():
        message = await channel.get()
        await websocket.send_text(message.decode())

2. Использование RabbitMQ:
   RabbitMQ — популярный брокер сообщений, который поддерживает различные шаблоны обмена сообщениями, включая Pub-Sub. Чтобы интегрировать RabbitMQ с FastAPI, вы можете использовать библиотеку aio_pika. Вот пример приложения FastAPI, которое демонстрирует, как использовать RabbitMQ для обмена сообщениями Pub-Sub:

import asyncio
from fastapi import FastAPI
import aio_pika
app = FastAPI()
async def publish_message(exchange, message):
    async with exchange.connect() as channel:
        await channel.default_exchange.publish(
            aio_pika.Message(body=message.encode()),
            routing_key="",
        )
@app.post("/publish/{exchange}")
async def publish(exchange: str, message: str):
    connection = await aio_pika.connect_robust("amqp://guest:guest@localhost/")
    async with connection:
        channel = await connection.channel()
        await channel.declare_exchange(exchange, auto_delete=True)
        await publish_message(channel.default_exchange, message)
        return {"status": "Message published"}
@app.websocket("/subscribe/{exchange}")
async def subscribe(exchange: str, websocket: WebSocket):
    connection = await aio_pika.connect_robust("amqp://guest:guest@localhost/")
    async with connection:
        channel = await connection.channel()
        await channel.declare_exchange(exchange, auto_delete=True)
        queue = await channel.declare_queue(exclusive=True)
        await queue.bind(exchange)
        async with queue.iterator() as queue_iter:
            async for message in queue_iter:
                await websocket.send_text(message.body.decode())
                await message.ack()

3. Использование Apache Kafka.
   Apache Kafka — это распределенная потоковая платформа, обеспечивающая надежные возможности обмена сообщениями Pub-Sub. Чтобы использовать Kafka с FastAPI, вы можете использовать библиотеку aiokafka. Вот пример приложения FastAPI, которое демонстрирует, как публиковать сообщения и подписываться на них с помощью Kafka:

import asyncio
from fastapi import FastAPI
from aiokafka import AIOKafkaProducer, AIOKafkaConsumer
app = FastAPI()
async def publish_message(producer, topic, message):
    await producer.send_and_wait(topic, message.encode())
@app.post("/publish/{topic}")
async def publish(topic: str, message: str):
    producer = AIOKafkaProducer(bootstrap_servers="localhost:9092")
    await producer.start()
    await publish_message(producer, topic, message)
    await producer.stop()
    return {"status": "Message published"}
@app.websocket("/subscribe/{topic}")
async def subscribe(topic: str, websocket: WebSocket):
    consumer = AIOKafkaConsumer(
        topic,
        bootstrap_servers="localhost:9092",
        group_id="my-group",
    )
    await consumer.start()
    async for message in consumer:
        await websocket.send_text(message.value.decode())

В этой статье мы рассмотрели различные методы реализации обмена сообщениями Pub-Sub с помощью FastAPI. Мы рассмотрели использование Redis Pub-Sub, RabbitMQ и Apache Kafka, предоставив примеры кода для каждого подхода. Используя эти шаблоны обмена сообщениями, вы можете улучшить масштабируемость и возможности работы в реальном времени ваших приложений FastAPI. Выберете ли вы Redis, RabbitMQ или Kafka, зависит от ваших конкретных требований и характеристик вашей системы.

Реализуя обмен сообщениями Pub-Sub с помощью FastAPI, вы можете создавать надежные и масштабируемые приложения, которые эффективно обрабатывают асинхронную связь между компонентами. Понимание этих методов позволит вам использовать возможности обмена сообщениями Pub-Sub в ваших проектах FastAPI, открывая новые возможности для обработки данных в реальном времени, архитектуры, управляемой событиями, и многого другого.