Kafkaイベントの非同期生成と利用

FastAPI runs on an asyncio event loop. If you publish or poll Kafka with a blocking client (like the standard kafka-python library), every network call freezes the entire loop, stalling all concurrent requests.

• aiokafka is a native asyncio Kafka client that never blocks the loop.
• Its send and getone operations are coroutines you await.
• This lets one worker handle thousands of in-flight requests while Kafka I/O is pending.

In this lesson you will wire an AIOKafkaProducer and an AIOKafkaConsumer into a FastAPI app the right way.

A producer maintains TCP connections and a background sender task. You must start() it once at app boot and stop() it on shutdown — never per request. The modern FastAPI way is the lifespan context manager.

• await producer.start() opens connections and the sender loop.
• await producer.stop() flushes pending batches and closes cleanly.
• Store the producer on app.state so routes can reach it.

Inside a route, grab the shared producer and await producer.send_and_wait(...). The send_and_wait call returns once the broker has acknowledged the record, giving you back-pressure and delivery confirmation.

• Kafka keys and values are bytes — encode JSON yourself or pass a serializer.
• The returned RecordMetadata tells you the partition and offset.
• Use the message key to guarantee per-entity ordering across partitions.

Instead of calling json.dumps(...).encode() on every send, you can hand aiokafka a value_serializer and key_serializer. The producer applies them automatically, so routes pass plain Python objects.

• value_serializer receives your object and must return bytes.
• This centralizes encoding and avoids repetition across routes.
• In production teams often swap JSON for Avro or Protobuf via a schema registry.

Two producer methods, two trade-offs:

• send() returns a future immediately and lets aiokafka batch records in the background — high throughput, but you do not yet know if delivery succeeded.
• send_and_wait() awaits the broker ack — slower per call, but gives you a result and surfaces errors right away.

For request handlers where the client needs confirmation, prefer send_and_wait. For fire-and-forget bulk emits, use send and optionally await the futures later.

The single most common mistake is mixing a synchronous client into the async app. Below is a standalone demo of why blocking the loop hurts: a blocking time.sleep inside a coroutine stalls everything, while asyncio.sleep yields control.

Run it and notice the awaited version lets both tasks overlap — exactly the behavior aiokafka gives you for real network I/O.

A FastAPI app serves HTTP, but a Kafka consumer must poll continuously. The clean pattern: start the consumer in lifespan and run its poll loop as an asyncio background task, cancelling it on shutdown.

• asyncio.create_task(...) launches the loop without blocking startup.
• On shutdown, cancel the task, then await consumer.stop().
• Always wrap the loop body so one bad message does not kill the consumer.

An AIOKafkaConsumer is an async iterator: async for msg in consumer awaits each new record. Each msg exposes topic, partition, offset, key, and value as bytes.

• Decode msg.value the same way you encoded it on the producer side.
• You can also pass a value_deserializer to the consumer constructor.
• msg.timestamp carries the broker or producer timestamp for latency metrics.

Auto-commit (the default) commits offsets on a timer, which can lose messages if the worker crashes after committing but before processing. For reliable processing, set enable_auto_commit=False and commit after you finish handling a message.

• Manual commit gives at-least-once semantics — a crash replays the last uncommitted message.
• Because messages can repeat, your handlers must be idempotent.
• Commit in small batches to balance throughput against replay cost.

Within a single partition Kafka preserves order, and an async-for loop processes those records sequentially. To scale, you have two levers:

• More partitions + more consumers in the same group_id — Kafka assigns partitions across instances automatically.
• Bounded concurrency per worker with a semaphore, when per-message work is I/O-heavy and strict ordering is not required.

If ordering per entity matters, keep that entity on one partition via a stable message key and process its partition serially.

A robust deployment cleans up so in-flight data is not lost:

• On shutdown, cancel the consumer task and await consumer.stop() — this commits offsets (if auto-commit) and leaves the group cleanly so rebalancing is fast.
• await producer.stop() flushes buffered records before closing.
• Wrap per-message handling in try/except and route poison messages to a dead-letter topic instead of crashing the loop.

Never call start()/stop() inside request handlers — that thrashes connections and breaks consumer group membership.

You need reliable processing where a worker crash must never silently drop an order event. Which consumer configuration best supports this?

You integrated Kafka into FastAPI without blocking the event loop:

• aiokafka provides awaitable producer and consumer clients native to asyncio.
• Manage the producer and consumer in the lifespan context: start() at boot, stop() at shutdown — never per request.
• send_and_wait confirms delivery; send maximizes throughput.
• Run the consumer poll loop as an asyncio background task and iterate with async for.
• Disable auto-commit and commit after processing for at-least-once delivery, keeping handlers idempotent.
• Scale with more partitions and consumers in a group, preserve per-entity order via the message key, and shut down gracefully with a dead-letter topic for poison messages.