aiokafka_async_producer_consumer_quickstart_example.py

1from aiokafka import AIOKafkaProducer, AIOKafkaConsumer
2import asyncio
3
4async def send_one():
5    producer = AIOKafkaProducer(
6        bootstrap_servers='localhost:9092')
7    # Get cluster layout and initial topic/partition guidance
8    await producer.start()
9    try:
10        # Produce message
11        await producer.send_and_wait("my_topic", b"Super message")
12    finally:
13        # Wait for all pending messages to be delivered or expire.
14        await producer.stop()
15
16async def consume():
17    consumer = AIOKafkaConsumer(
18        'my_topic',
19        bootstrap_servers='localhost:9092',
20        group_id="my-group")
21    # Get cluster layout and join group `my-group`
22    await consumer.start()
23    try:
24        # Consume messages
25        async for msg in consumer:
26            print("consumed: ", msg.topic, msg.partition, msg.offset,
27                  msg.key, msg.value, msg.timestamp)
28    finally:
29        # Will leave consumer group; perform autocommit if enabled.
30        await consumer.stop()
31
32async def main():
33    # Note: In a real application these would likely run in separate processes or tasks
34    await send_one()
35    await consume()
36
37if __name__ == "__main__":
38    asyncio.run(main())