Build một hệ thống Kafka-Based FastAPI Microservices với Real-Time Event Streaming

# microservice-a/app/main.py

import asyncio

from fastapi import FastAPI

from pydantic import BaseModel

from aiokafka import AIOKafkaProducer

from aiokafka.errors import KafkaConnectionError

app = FastAPI()

KAFKA_TOPIC = "test-topic"

KAFKA_BOOTSTRAP_SERVERS = "kafka:9092"

producer: AIOKafkaProducer | None = None

class Message(BaseModel):

content: str

@app.on_event("startup")

async def startup_event():

global producer

producer = AIOKafkaProducer(bootstrap_servers=KAFKA_BOOTSTRAP_SERVERS)

while True:

try:

await producer.start()

print("Kafka producer connected")

break

except KafkaConnectionError:

print("Kafka producer not available. Retrying in 5 seconds...")

await asyncio.sleep(5)

@app.on_event("shutdown")

async def shutdown_event():

global producer

if producer:

await producer.stop()

print("Kafka producer stopped")

@app.post("/produce")

async def produce_message(message: Message):

if producer is None:

return {"error": "Kafka producer is not started"}

await producer.send_and_wait(KAFKA_TOPIC, message.content.encode("utf-8"))

return {"status": "message sent", "content": message.content}

# microservice-b/app/main.py

from fastapi import FastAPI

import asyncio

from aiokafka import AIOKafkaConsumer

from aiokafka.errors import KafkaConnectionError

app = FastAPI()

KAFKA_TOPIC = "test-topic"

KAFKA_BOOTSTRAP_SERVERS = "kafka:9092"

async def consume():

print(f"Entering in consume")

consumer = AIOKafkaConsumer(

KAFKA_TOPIC,

bootstrap_servers=KAFKA_BOOTSTRAP_SERVERS,

group_id="my-fastapi-consumer-group",

auto_offset_reset="earliest",

)

while True:

try:

await consumer.start()

print("Kafka consumer connected and listening...")

break

except KafkaConnectionError:

print("Kafka not available. Retrying in 5 seconds...")

await asyncio.sleep(5)

while True:

try:

print("Waiting for messages..")

async for msg in consumer:

print(f"Message received: {msg.value.decode('utf-8')}")

except Exception as e:

print(f"Error in consumer loop: {e}")

await asyncio.sleep(5)

@app.on_event("startup")

async def startup_event():

print("Startup event launched...")

await asyncio.sleep(10)

asyncio.create_task(consume())

@app.get("/")

async def root():

return {"message": "Microservice B is running!"}

# Variables

PROJECT_NAME=fastapi-microservices-kafka

COMPOSE=docker-compose

PY=python3

# Start development environment with hot reload

up:

@echo "Starting development environment..."

$(COMPOSE) up --build

# Stop and remove all containers, volumes, and networks

down:

@echo "Stopping and cleaning up containers and volumes..."

$(COMPOSE) down -v

# Stop containers without removing volumes

stop:

$(COMPOSE) down

# Rebuild images without using cache

rebuild:

$(COMPOSE) build --no-cache

# Send a test message to the Kafka producer

test-produce:

curl -X POST http://localhost:8001/produce \

-H "Content-Type: application/json" \

-d '{"content": "Test message from Makefile"}'

# View logs for the consumer microservice

logs-b:

$(COMPOSE) logs -f microservice-b

# View logs for the producer microservice

logs-a:

$(COMPOSE) logs -f microservice-a

# Clean up unused Docker data

clean:

docker system prune -f

# Open a shell in the producer container

sh-a:

$(COMPOSE) exec microservice-a sh

# Open a shell in the consumer container

sh-b:

$(COMPOSE) exec microservice-b sh

# Start production environment

up-prod:

@echo "Starting production environment..."

docker-compose -f docker-compose.yml -f docker-compose.prod.yml up --build -d

curl -X POST http://localhost:8000/produce \

-H "Content-Type: application/json" \

-d '{"content": "Hello Kafka from FastAPI!"}'

make logs-b

Real-Time Use Cases