Apache Kafka（分布式流处理平台）

# 概念说明：Topic 和 Partition 的关系
# 假设创建一个 3 分区的 Topic
topic_config = {
    'name': 'agent_events',       # 主题名
    'num_partitions': 3,          # 3 个分区 → 最多 3 个消费者并行
    'replication_factor': 2       # 每个分区 2 个副本（1 Leader + 1 Follower）
}

# 消息路由示例：
# key="agent_001" → hash("agent_001") % 3 = 1 → 进入 Partition 1
# key="agent_002" → hash("agent_002") % 3 = 0 → 进入 Partition 0
# 同一个 agent 的所有事件进入同一分区，保证时间顺序

# Python 客户端（推荐 confluent-kafka，性能优于 kafka-python）
pip install confluent-kafka==2.6.1

# 或者使用 kafka-python（纯 Python 实现，安装更简单）
pip install kafka-python==2.0.2

# 使用 KRaft 模式（Kafka 3.3+ 不再需要 ZooKeeper）
docker run -d --name kafka \
  -p 9092:9092 \
  -e KAFKA_NODE_ID=1 \
  -e KAFKA_PROCESS_ROLES=broker,controller \
  -e KAFKA_LISTENERS=PLAINTEXT://0.0.0.0:9092,CONTROLLER://0.0.0.0:9093 \
  -e KAFKA_ADVERTISED_LISTENERS=PLAINTEXT://localhost:9092 \
  -e KAFKA_CONTROLLER_QUORUM_VOTERS=1@localhost:9093 \
  -e KAFKA_CONTROLLER_LISTENER_NAMES=CONTROLLER \
  -e KAFKA_LISTENER_SECURITY_PROTOCOL_MAP=CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT \
  -e CLUSTER_ID=MkU3OEVBNTcwNTJENDM2Qk \
  apache/kafka:3.8.0

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
Kafka 最小可运行示例：生产者发送 5 条消息，消费者读取并打印。
前提：本地 9092 端口已运行 Kafka 服务。
"""

from kafka import KafkaProducer, KafkaConsumer
from kafka.errors import KafkaError
import json
import threading
import time

TOPIC = "demo_topic"
SERVERS = ["localhost:9092"]

# ---- 生产者：发送 5 条消息 ----
def produce():
    producer = KafkaProducer(
        bootstrap_servers=SERVERS,
        value_serializer=lambda v: json.dumps(v, ensure_ascii=False).encode("utf-8"),
        acks="all",
    )

    for i in range(5):
        msg = {"id": i, "content": f"第 {i} 条消息"}
        future = producer.send(TOPIC, value=msg)
        metadata = future.get(timeout=10)
        print(f"[生产者] 发送成功 -> Partition {metadata.partition}, Offset {metadata.offset}")

    producer.flush()
    producer.close()

# ---- 消费者：读取消息 ----
def consume():
    time.sleep(2)  # 等待生产者先发几条
    consumer = KafkaConsumer(
        TOPIC,
        bootstrap_servers=SERVERS,
        group_id="demo_group",
        auto_offset_reset="earliest",
        value_deserializer=lambda m: json.loads(m.decode("utf-8")),
    )

    count = 0
    for message in consumer:
        print(f"[消费者] 收到: {message.value} (Partition={message.partition}, Offset={message.offset})")
        count += 1
        if count >= 5:
            break

    consumer.close()

# ---- 运行 ----
if __name__ == "__main__":
    t1 = threading.Thread(target=produce)
    t2 = threading.Thread(target=consume)
    t1.start()
    t2.start()
    t1.join()
    t2.join()
    print("演示完成")

[生产者] 发送成功 -> Partition 0, Offset 0
[生产者] 发送成功 -> Partition 0, Offset 1
[生产者] 发送成功 -> Partition 0, Offset 2
[生产者] 发送成功 -> Partition 0, Offset 3
[生产者] 发送成功 -> Partition 0, Offset 4
[消费者] 收到: {'id': 0, 'content': '第 0 条消息'} (Partition=0, Offset=0)
[消费者] 收到: {'id': 1, 'content': '第 1 条消息'} (Partition=0, Offset=1)
[消费者] 收到: {'id': 2, 'content': '第 2 条消息'} (Partition=0, Offset=2)
[消费者] 收到: {'id': 3, 'content': '第 3 条消息'} (Partition=0, Offset=3)
[消费者] 收到: {'id': 4, 'content': '第 4 条消息'} (Partition=0, Offset=4)
演示完成