Tự viết Producer / Consumer

Cả hai đều dùng Kafka, nhưng giải quyết hai bài toán khác nhau. Hãy chọn đúng công cụ ngay từ đầu:

Hai thư viện Python phổ biến để nói chuyện với Kafka:

Bài này dùng confluent-kafka. Cài bằng pip:

# cài client chính thức của Confluent
pip install confluent-kafka

Tham số quan trọng nhất là bootstrap.servers — địa chỉ broker để client kết nối lần đầu:

# từ máy host (chạy script Python ngoài Docker)
bootstrap.servers = "localhost:29092"

# từ trong cùng Docker network (vd container khác trong cdc-net)
bootstrap.servers = "kafka:9092"

Producer chịu trách nhiệm gửi message lên topic. Dưới đây là một producer hoàn chỉnh gửi vài bản ghi JSON vào topic demo:

# producer.py
import json
from confluent_kafka import Producer

# cấu hình kết nối broker
conf = {
    "bootstrap.servers": "localhost:29092",
    "acks": "all",              # chờ mọi replica xác nhận (bền nhất)
    "enable.idempotence": True,   # tránh ghi trùng khi retry
    "linger.ms": 10,             # gom message ~10ms để batch
    "batch.size": 32768,         # kích thước batch (bytes)
}

producer = Producer(conf)

# callback báo kết quả gửi (thành công / thất bại)
def delivery_report(err, msg):
    if err is not None:
        print(f"Gửi THẤT BẠI: {err}")
    else:
        print(f"OK -> topic={msg.topic()} "
              f"partition={msg.partition()} offset={msg.offset()}")

orders = [
    {"id": 1, "user": "an",  "amount": 120},
    {"id": 2, "user": "binh", "amount": 340},
    {"id": 3, "user": "an",  "amount": 90},
]

for o in orders:
    producer.produce(
        topic="demo",
        key=str(o["user"]).encode("utf-8"),     # key quyết định partition
        value=json.dumps(o).encode("utf-8"),       # value serialize sang JSON bytes
        callback=delivery_report,
    )
    producer.poll(0)   # gọi callback cho các message đã gửi xong

producer.flush()        # chặn cho tới khi gửi hết buffer

Các điểm cốt lõi của Producer:

• Key quyết định partition — message cùng key luôn rơi vào cùng một partition, nhờ đó giữ đúng thứ tự cho các message liên quan (vd cùng một user). Không truyền key thì Kafka chia đều round-robin.
• Serialization — Kafka chỉ hiểu bytes. Value thường được mã hóa JSON: json.dumps(obj).encode("utf-8"). Production lớn hay dùng Avro/Protobuf + Schema Registry để có schema chặt chẽ.
• poll() & flush() — produce() là bất đồng bộ, chỉ đẩy vào buffer. poll(0) kích hoạt các callback đã sẵn sàng; flush() chặn cho tới khi mọi message được gửi đi (luôn gọi trước khi thoát).
• Idempotent producer — bật enable.idempotence=True để Kafka tự khử trùng khi producer retry, tránh ghi 2 lần cùng một message.
• Batching — linger.ms cho phép gom message trong khoảng thời gian ngắn, batch.size giới hạn kích thước batch; tăng throughput, đổi lại độ trễ nhỏ.

Tham số acks đánh đổi giữa độ bền và tốc độ:

Consumer đăng ký (subscribe) một hay nhiều topic rồi lặp poll để lấy message. Đây là consumer hoàn chỉnh đọc topic demo với commit thủ công:

# consumer.py
import json
from confluent_kafka import Consumer, KafkaException

conf = {
    "bootstrap.servers": "localhost:29092",
    "group.id": "my-group",          # tên consumer group
    "auto.offset.reset": "earliest",  # chưa có offset thì đọc từ đầu topic
    "enable.auto.commit": False,      # tự commit để xử lý xong mới đánh dấu
}

consumer = Consumer(conf)
consumer.subscribe(["demo"])

try:
    while True:
        msg = consumer.poll(1.0)   # chờ tối đa 1 giây
        if msg is None:
            continue
        if msg.error():
            raise KafkaException(msg.error())

        # deserialize JSON về dict
        data = json.loads(msg.value().decode("utf-8"))
        key = msg.key().decode("utf-8") if msg.key() else None

        print(f"key={key} value={data} "
              f"partition={msg.partition()} offset={msg.offset()}")

        # ... xử lý nghiệp vụ tại đây ...

        consumer.commit(msg)   # xử lý xong mới commit offset
except KeyboardInterrupt:
    pass
finally:
    consumer.close()        # rời group gọn gàng, kích hoạt rebalance

Các khái niệm then chốt của Consumer:

• Consumer group & phân chia partition — các consumer cùng group.id chia nhau các partition của topic. Mỗi partition chỉ do một consumer trong group đọc, nhờ đó scale ngang: thêm consumer (tối đa = số partition) để tăng throughput.
• Offset & commit — offset đánh dấu vị trí đã đọc tới. Auto commit (enable.auto.commit=True) tiện nhưng có thể commit trước khi xử lý xong. Commit thủ công sau khi xử lý thành công an toàn hơn (đảm bảo at-least-once).
• auto.offset.reset — khi group chưa có offset lưu: earliest đọc từ đầu topic, latest chỉ đọc message mới đến.
• Rebalance — khi thêm/bớt consumer hoặc một consumer chết, Kafka phân bổ lại partition cho các consumer còn lại. Gọi close() khi tắt để rời group sạch sẽ, rebalance nhanh hơn.
• Deserialization — đảo ngược của producer: json.loads(msg.value().decode("utf-8")) để biến bytes JSON về dict Python.

Thứ tự giữa xử lý và commit offset quyết định bạn nhận được mức đảm bảo nào:

Thực tế đa số dùng at-least-once (commit sau khi xử lý) rồi làm cho việc xử lý idempotent để chịu được message trùng: ví dụ dùng khóa nghiệp vụ (order id) với UPSERT hoặc kiểm tra "đã xử lý chưa" trước khi tác động — kết quả cuối cùng giống như chỉ xử lý đúng một lần.

Một consumer chạy production cần bao bọc xử lý lỗi cẩn thận: bắt exception cho từng message, retry với lỗi tạm thời, đẩy message hỏng sang Dead Letter Topic, và tắt êm khi nhận tín hiệu dừng.

# consumer có DLQ tự xây + commit sau khi xử lý
import json, signal
from confluent_kafka import Producer, Consumer

running = True
def stop(*_):           # graceful shutdown
    global running
    running = False
signal.signal(signal.SIGINT, stop)
signal.signal(signal.SIGTERM, stop)

base = {"bootstrap.servers": "localhost:29092"}
consumer = Consumer({**base, "group.id": "my-group",
                     "auto.offset.reset": "earliest",
                     "enable.auto.commit": False})
dlq = Producer(base)
consumer.subscribe(["demo"])

try:
    while running:
        msg = consumer.poll(1.0)
        if msg is None or msg.error():
            continue
        try:
            data = json.loads(msg.value().decode("utf-8"))
            handle(data)                # nghiệp vụ của bạn
            consumer.commit(msg)         # chỉ commit khi thành công
        except Exception as e:
            # gửi message lỗi sang Dead Letter Topic để điều tra
            dlq.produce("demo.DLT", key=msg.key(), value=msg.value())
            dlq.flush()
            consumer.commit(msg)         # bỏ qua, đi tiếp
            print(f"Đẩy DLT: {e}")
finally:
    consumer.close()

• try/except mỗi message — một message hỏng không được làm sập cả vòng lặp.
• Retry — với lỗi tạm thời (mạng, DB chập chờn) hãy retry vài lần có backoff trước khi bỏ cuộc.
• DLQ tự xây — message không xử lý được thì gửi sang topic .DLT rồi vẫn commit để pipeline đi tiếp; điều tra DLT sau.
• Commit thủ công sau xử lý — chỉ đánh dấu đã đọc khi chắc chắn xử lý xong (hoặc đã đẩy DLT).
• Graceful shutdown — bắt SIGINT/SIGTERM, thoát vòng lặp và gọi close() để commit nốt và rời group.

• Bật cdc-lab để có Kafka chạy ở localhost:29092 (broker) và Kafka UI ở localhost:8080.
• Cài client: pip install confluent-kafka.
• Chạy python producer.py để gửi vài message vào topic demo (topic tự tạo nếu broker bật auto-create).
• Chạy python consumer.py ở cửa sổ khác để đọc message ra màn hình.
• Mở Kafka UI tại localhost:8080 để xem topic demo, các message, và consumer lag của group my-group.