Apache Kafka 完全指南 2026：KRaft 架构 + 核心模型 + 性能调优 + Python/Go 实战

1
import asyncio
2
import json
3
from aiokafka import AIOKafkaProducer
4
from pydantic import BaseModel
5

6
class OrderEvent(BaseModel):
7
    order_id: str
8
    user_id: int
9
    amount: float
10
    timestamp: int
11

12
async def send_orders():
13
    producer = AIOKafkaProducer(
14
        bootstrap_servers='localhost:9092',
15
        # 批量聚合与压缩配置，最大化吞吐
16
        linger_ms=10,                      # 攒批等待时间
17
        compression_type="zstd",           # 现代高效压缩
18
        acks="all",                        # 保证最强持久性
19
        enable_idempotence=True            # 启用生产者幂等性
20
    )
21
    await producer.start()
22
    try:
23
        for i in range(1000):
24
            event = OrderEvent(
25
                order_id=f"ord_{1000 + i}",
26
                user_id=100 + (i % 10),
27
                amount=99.5 + i,
28
                timestamp=1723700000 + i
29
            )
30
            # 使用 user_id 作为 key，保证同一用户的订单进入同一分区，严格有序
31
            key = str(event.user_id).encode("utf-8")
32
            val = event.model_dump_json().encode("utf-8")
33

34
            await producer.send("order_events", key=key, value=val)
35
        print("1000 events sent successfully.")
36
    finally:
37
        await producer.stop()
38

39
if __name__ == "__main__":
40
    asyncio.run(send_orders())

1
import asyncio
2
from aiokafka import AIOKafkaConsumer
3

4
async def consume_orders():
5
    consumer = AIOKafkaConsumer(
6
        "order_events",
7
        bootstrap_servers='localhost:9092',
8
        group_id="order-processor-group",
9
        auto_offset_reset="earliest",      # 从最早位点开始
10
        enable_auto_commit=False,          # 关闭自动提交，手动提交避免丢失
11
        max_poll_records=200               # 每次批量拉取条数
12
    )
13
    await consumer.start()
14
    try:
15
        while True:
16
            # 批量拉取数据
17
            batch = await consumer.getmany(timeout_ms=1000, max_records=200)
18
            for tp, messages in batch.items():
19
                if not messages:
20
                    continue
21
                print(f"Processing {len(messages)} messages from partition {tp.partition}")
22
                for msg in messages:
23
                    # 业务处理逻辑
24
                    pass
25

26
                # 处理成功后手动提交位点
27
                await consumer.commit()
28
    finally:
29
        await consumer.stop()
30

31
if __name__ == "__main__":
32
    asyncio.run(consume_orders())

1
package main
2

3
import (
4
  "encoding/json"
5
  "fmt"
6
  "github.com/confluentinc/confluent-kafka-go/v2/kafka"
7
)
8

9
type Event struct {
10
  UserID string `json:"user_id"`
11
  Action string `json:"action"`
12
}
13

14
func main() {
15
  p, err := kafka.NewProducer(&kafka.ConfigMap{
16
    "bootstrap.servers": "localhost:9092",
17
    "acks":              "all",
18
    "enable.idempotence": true,
19
    "compression.type":  "lz4",
20
    "linger.ms":         5,
21
  })
22
  if err != nil {
23
    panic(err)
24
  }
25
  defer p.Close()
26

27
  topic := "user_actions"
28
  event := Event{UserID: "u_888", Action: "click_buy"}
29
  payload, _ := json.Marshal(event)
30

31
  // 异步发送
32
  err = p.Produce(&kafka.Message{
33
    TopicPartition: kafka.TopicPartition{Topic: &topic, Partition: kafka.PartitionAny},
34
    Key:            []byte(event.UserID),
35
    Value:          payload,
36
  }, nil)
37

38
  // 等待所有在途消息送达
39
  p.Flush(5000)
40
  fmt.Println("Message sent.")
41
}