Giới thiệu

AIOKafkaProducer là client bất đồng bộ dùng để xuất bản bản ghi lên cụm Kafka. Các ví dụ dưới đây được viết cho aiokafka 0.8+ và Python 3.8 trở lên, sử dụng asyncio.run thay vì truyền loop trực tiếp.

Gửi một tin nhắn đơn

Khi tạo producer, gọi start(), gửi tin bằng send_and_wait(), rồi dừng producer trong khối finally.

import asyncio
import datetime
from aiokafka import AIOKafkaProducer

BOOTSTRAP = "192.168.1.3:9092"
TOPIC = "my_topic"

def now(fmt="%Y-%m-%d %H:%M:%S"):
    return datetime.datetime.now().strftime(fmt)

async def publish_single():
    producer = AIOKafkaProducer(bootstrap_servers=BOOTSTRAP)
    await producer.start()
    try:
        payload = f"now: {now()}".encode()
        await producer.send_and_wait(TOPIC, value=payload)
        print("Gửi tin nhắn thành công")
    except Exception as exc:
        print(f"Gửi tin nhắn thất bại: {exc}")
    finally:
        await producer.stop()

if __name__ == "__main__":
    asyncio.run(publish_single())

Gửi nhiều tin nhắn theo thứ tự

Khi gọi await trong vòng lặp, mỗi tin nhắn phải ch tin trước hoàn thành mới tiếp tục, do đó các bản ghi được gửi tuần tự.

import asyncio
from aiokafka import AIOKafkaProducer

BOOTSTRAP = "192.168.1.3:9092"
TOPIC = "my_topic"

class KafkaPublisher:
    def __init__(self, bootstrap_servers):
        self.producer = AIOKafkaProducer(bootstrap_servers=bootstrap_servers)

    async def __aenter__(self):
        await self.producer.start()
        return self

    async def __aexit__(self, exc_type, exc, tb):
        await self.producer.stop()

    async def publish(self, topic, message):
        await self.producer.send_and_wait(topic, message.encode())

async def sequential_send(messages):
    async with KafkaPublisher(BOOTSTRAP) as pub:
        for msg in messages:
            await pub.publish(TOPIC, msg)
            print(f"Đã gửi: {msg}")

if __name__ == "__main__":
    messages = [f"seq-{i}" for i in range(20)]
    asyncio.run(sequential_send(messages))

Gửi nhiều tin nhắn song song

Tạo danh sách các coroutine rồi truyền vào asyncio.gather. Các tác vụ chạy đồng thời nên thứ tự gửi không còn được đảm bảo.

import asyncio
from aiokafka import AIOKafkaProducer

BOOTSTRAP = "192.168.1.3:9092"
TOPIC = "my_topic"

class KafkaPublisher:
    def __init__(self, bootstrap_servers):
        self.producer = AIOKafkaProducer(bootstrap_servers=bootstrap_servers)

    async def __aenter__(self):
        await self.producer.start()
        return self

    async def __aexit__(self, exc_type, exc, tb):
        await self.producer.stop()

    async def publish(self, topic, message):
        return await self.producer.send_and_wait(topic, message.encode())

async def concurrent_send(messages):
    async with KafkaPublisher(BOOTSTRAP) as pub:
        tasks = [pub.publish(TOPIC, msg) for msg in messages]
        results = await asyncio.gather(*tasks, return_exceptions=True)

        for msg, result in zip(messages, results):
            if isinstance(result, Exception):
                print(f"Gửi thất bại [{msg}]: {result}")
            else:
                print(f"Đã gửi: {msg}")

if __name__ == "__main__":
    messages = [f"par-{i}" for i in range(20)]
    asyncio.run(concurrent_send(messages))

Gửi tin nhắn từ dòng lệnh

Đoạn mã dưi đây nhận đầu vào từ bàn phím và gửi lên Kafka cho đến khi người dùng nhập quit. Hàm input được chạy trong executor để không chặn vòng lặp sự kiện.

import asyncio
from aiokafka import AIOKafkaProducer

BOOTSTRAP = "192.168.1.3:9092"
TOPIC = "my_topic"

async def read_line(prompt="> "):
    loop = asyncio.get_event_loop()
    return await loop.run_in_executor(None, input, prompt)

async def interactive_send():
    producer = AIOKafkaProducer(bootstrap_servers=BOOTSTRAP)
    await producer.start()
    try:
        print("Nhập tin nhắn cần gửi (gõ 'quit' để thoát):")
        while True:
            line = await read_line("message: ")
            text = line.strip()
            if text.lower() == "quit":
                print("Kết thúc")
                break
            await producer.send_and_wait(TOPIC, text.encode())
            print("  -> đã gửi")
    finally:
        await producer.stop()

if __name__ == "__main__":
    asyncio.run(interactive_send())