3. 数据源与消息队列：用户行为日志生成器设计、Apache Kafka核心概念、Kafka集群搭建（Docker）、Topic设计与分区策略

好，咱们进入第三章。这一章是整个实时推荐平台的「水源」和「血管」。数据源怎么来？消息队列怎么扛住千万级流量？我见过不少团队，算法模型再牛，数据管道一堵，全白搭。所以这一章，咱们把地基打扎实。

3.1 用户行为日志生成器设计

做实时推荐，第一步就是要有数据。但项目初期哪有真实用户？所以咱们自己写一个日志生成器。说白了，就是模拟用户在你网站上的各种操作：浏览商品、点击详情、加入购物车、下单购买。

我个人习惯用 Python 来写这类模拟器，因为快。你想想看，一个 while 循环，配合 random 库，就能模拟出几万用户的访问模式。

来看一个简化版的生成器核心代码：

import json, time, random, uuid
from datetime import datetime

# 模拟用户画像
USER_PROFILES = {
    "user_001": {"interests": ["手机", "电脑", "耳机"]},
    "user_002": {"interests": ["图书", "文具", "课程"]},
    "user_003": {"interests": ["母婴", "玩具", "童装"]},
}

ACTIONS = ["view", "click", "add_cart", "purchase"]
ITEMS = {
    "手机": ["iphone_15", "xiaomi_14", "huawei_p70"],
    "电脑": ["macbook_pro", "thinkpad_x1", "rog_zephyrus"],
    "图书": ["python_cookbook", "kafka_definitive", "design_data_intensive"],
    "母婴": ["baby_car_seat", "diaper_pack", "stroller_x"],
}

def generate_log():
    user_id = random.choice(list(USER_PROFILES.keys()))
    profile = USER_PROFILES[user_id]
    category = random.choice(profile["interests"])
    item = random.choice(ITEMS[category])
    action = random.choice(ACTIONS)
    
    log = {
        "user_id": user_id,
        "item_id": item,
        "category": category,
        "action": action,
        "timestamp": datetime.utcnow().isoformat() + "Z",
        "session_id": str(uuid.uuid4())[:8]
    }
    return json.dumps(log)

# 模拟持续发送
if __name__ == "__main__":
    while True:
        print(generate_log())
        time.sleep(random.uniform(0.5, 2.0))  # 随机间隔

这段代码我实际项目里改过很多版。最开始是固定间隔，结果下游 Kafka 消费端压力测试完全不准。后来改成随机间隔 + 突发流量模拟，才更贴近真实场景。

3.2 Apache Kafka 核心概念

日志生成出来了，怎么传给下游？直接 HTTP 调用？并发一高后端就挂了。这时候就需要消息队列。Kafka 是目前实时数据管道的事实标准。

Kafka 的核心概念其实不多，但每个都很关键。我把它总结成一张表：

概念	一句话解释	我见过的坑
Producer	发送消息的客户端	没配置 acks=all，丢过消息
Consumer	消费消息的客户端	消费组 rebalance 导致重复消费
Broker	Kafka 服务器节点	磁盘空间打满，整个集群 hang 住
Topic	消息的逻辑分类	Topic 数量过多，文件句柄耗尽
Partition	Topic 的分片，并行度单位	分区数设置后不可减少
Offset	消息在分区内的序号	手动提交 offset 容易忘
ISR	与 Leader 保持同步的副本集合	ISR 收缩导致写入延迟飙升

这里重点说说 Partition。Kafka 的并行度就靠它。一个 Topic 有 N 个分区，最多就能被 N 个 Consumer 同时消费。分区数太少，消费能力上不去；分区数太多，rebalance 和文件管理开销又大。

3.3 Kafka 集群搭建（Docker）

搭建 Kafka 集群，传统方式要装 JDK、解压 tar 包、改配置、启动 Zookeeper……太麻烦了。用 Docker Compose，10 分钟搞定。

这是我常用的 docker-compose.yml 配置：

version: '3.8'
services:
  zookeeper:
    image: confluentinc/cp-zookeeper:7.5.0
    environment:
      ZOOKEEPER_CLIENT_PORT: 2181
      ZOOKEEPER_TICK_TIME: 2000

  kafka1:
    image: confluentinc/cp-kafka:7.5.0
    depends_on:
      - zookeeper
    ports:
      - "9092:9092"
    environment:
      KAFKA_BROKER_ID: 1
      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://localhost:9092
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 3

  kafka2:
    image: confluentinc/cp-kafka:7.5.0
    depends_on:
      - zookeeper
    ports:
      - "9093:9092"
    environment:
      KAFKA_BROKER_ID: 2
      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://localhost:9093
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 3

  kafka3:
    image: confluentinc/cp-kafka:7.5.0
    depends_on:
      - zookeeper
    ports:
      - "9094:9092"
    environment:
      KAFKA_BROKER_ID: 3
      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://localhost:9094
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 3

启动命令就一行：

docker-compose up -d

然后验证集群状态：

# 查看 broker 列表
docker exec -it kafka1 kafka-broker-api-versions --bootstrap-server localhost:9092

# 创建测试 topic
docker exec -it kafka1 kafka-topics --create \
  --topic test_topic \
  --bootstrap-server localhost:9092 \
  --partitions 3 \
  --replication-factor 3

# 查看 topic 详情
docker exec -it kafka1 kafka-topics --describe \
  --topic test_topic \
  --bootstrap-server localhost:9092

3.4 Topic 设计与分区策略

Topic 设计得好不好，直接决定了后续数据处理的复杂度。我见过最惨的案例：一个团队把所有用户行为都塞进一个 Topic，结果下游不同业务线互相干扰，一个消费组出问题，全链路瘫痪。

对于实时推荐平台，我建议这样设计：

Topic 名称	用途	分区数	保留策略
user_behavior	用户原始行为日志	6	7 天
item_feature	物品特征变更	3	紧凑保留
recommend_result	推荐结果输出	3	1 天
model_train_event	模型训练触发事件	1	1 天

分区策略上，我推荐用 用户 ID 哈希分区。为什么？因为同一个用户的行为必须有序。如果用户 A 的「浏览」和「购买」被分到不同分区，消费端就无法保证顺序。

Producer 端代码示例：

from kafka import KafkaProducer
import json

producer = KafkaProducer(
    bootstrap_servers=['localhost:9092', 'localhost:9093', 'localhost:9094'],
    value_serializer=lambda v: json.dumps(v).encode('utf-8'),
    # 自定义分区器：按用户 ID 哈希
    partitioner=lambda key_bytes, all_partitions, available: (
        hash(key_bytes) % len(all_partitions)
    ),
    acks='all',  # 等待所有副本确认
    retries=3
)

# 发送消息
log_data = {"user_id": "user_001", "action": "purchase", "item": "iphone_15"}
producer.send('user_behavior', key=b'user_001', value=log_data)
producer.flush()