Apache Kafka是什么？分布式事件流处理平台详解

Order Service → Inventory (HTTP)
             → Notification (HTTP)
             → Analytics (HTTP)
             → Fraud (HTTP)
             → Recommendation (HTTP)

Order Service → [order-placed Topic] → Inventory Service
                                     → Notification Service
                                     → Analytics Service
                                     → Fraud Service
                                     → Recommendation Service

Topic: order-placed（3个Partition）
│
├─ Consumer Group "inventory-svc"   → Consumer A (P0)，Consumer B (P1, P2)
└─ Consumer Group "analytics-svc"  → Consumer X (P0, P1, P2)

# 保留消息7天
log.retention.hours=168

# 或按大小限制
log.retention.bytes=10737418240  # 每个Partition 10 GB

 1from kafka import KafkaProducer, KafkaConsumer
 2import json
 3
 4# Producer：向Topic写入消息
 5producer = KafkaProducer(
 6    bootstrap_servers=['localhost:9092'],
 7    value_serializer=lambda v: json.dumps(v).encode('utf-8'),
 8    # 等待所有同步副本确认，确保不丢消息
 9    acks='all',
10    retries=3
11)
12
13# 发送下单事件
14order_event = {
15    "event": "order_placed",
16    "order_id": "ORD-2026-001",
17    "user_id": 42,
18    "total": 199.00
19}
20
21# Partition Key = user_id，确保同一用户的事件进入同一Partition
22future = producer.send(
23    topic='order-placed',
24    key=str(order_event['user_id']).encode('utf-8'),
25    value=order_event
26)
27record_metadata = future.get(timeout=10)
28print(f"已发送 → partition={record_metadata.partition}, offset={record_metadata.offset}")
29
30producer.flush()
31producer.close()
32
33# Consumer Group：从Topic消费消息
34consumer = KafkaConsumer(
35    'order-placed',
36    bootstrap_servers=['localhost:9092'],
37    group_id='inventory-service',  # Consumer Group ID
38    auto_offset_reset='earliest',  # 没有历史offset时从头开始读取
39    value_deserializer=lambda m: json.loads(m.decode('utf-8')),
40    enable_auto_commit=False  # 关闭自动提交，手动控制offset
41)
42
43print("库存服务正在监听...")
44try:
45    for message in consumer:
46        event = message.value
47        print(f"收到：order_id={event['order_id']}, user={event['user_id']}")
48
49        # 在此处理业务逻辑（扣减库存等）
50        # ...
51
52        # 成功处理后手动提交offset
53        consumer.commit()
54finally:
55    consumer.close()

 1const { Kafka } = require('kafkajs');
 2
 3const kafka = new Kafka({
 4  clientId: 'my-app',
 5  brokers: ['localhost:9092'],
 6});
 7
 8// ── Producer ──────────────────────────────────────────────
 9async function runProducer() {
10  const producer = kafka.producer();
11  await producer.connect();
12
13  // 批量发送消息到Topic
14  await producer.send({
15    topic: 'order-placed',
16    messages: [
17      {
18        key: '42',                               // Partition Key = user_id
19        value: JSON.stringify({
20          event: 'order_placed',
21          orderId: 'ORD-2026-002',
22          userId: 42,
23          total: 299.00,
24        }),
25      },
26      {
27        key: '99',
28        value: JSON.stringify({
29          event: 'order_placed',
30          orderId: 'ORD-2026-003',
31          userId: 99,
32          total: 89.50,
33        }),
34      },
35    ],
36  });
37
38  console.log('已向Kafka发送2条下单事件');
39  await producer.disconnect();
40}
41
42// ── Consumer Group ────────────────────────────────────────
43async function runConsumer() {
44  const consumer = kafka.consumer({ groupId: 'notification-service' });
45  await consumer.connect();
46
47  // 订阅Topic
48  await consumer.subscribe({ topic: 'order-placed', fromBeginning: false });
49
50  await consumer.run({
51    // 每收到一条消息都会调用eachMessage
52    eachMessage: async ({ topic, partition, message }) => {
53      const event = JSON.parse(message.value.toString());
54      console.log(`[P${partition}] orderId=${event.orderId}, userId=${event.userId}`);
55
56      // 发送确认邮件或推送通知
57      // await notificationSvc.send(event);
58    },
59  });
60}
61
62runProducer().catch(console.error);
63runConsumer().catch(console.error);

 1// 示例：Debezium MySQL Source Connector配置
 2{
 3  "name": "mysql-orders-connector",
 4  "config": {
 5    "connector.class": "io.debezium.connector.mysql.MySqlConnector",
 6    "database.hostname": "mysql-host",
 7    "database.port": "3306",
 8    "database.user": "debezium",
 9    "database.password": "secret",
10    "database.server.name": "mydb",
11    "table.include.list": "shop.orders"
12  }
13}