📚 课程目录第2章数据源接入

数据源接入：Kafka 基础与安装配置、生产者与消费者 API、模拟数据生成器开发

数据管道的第一公里，就是数据源接入。你想想看，如果源头的数据都接不稳，后面流批一体做得再花哨也是白搭。我个人习惯用 Kafka 作为数据管道的中枢神经系统——它扛得住高峰流量，又能把生产和消费解耦得干干净净。

这一章，咱们就手把手把 Kafka 环境搭起来，把生产者和消费者的 API 调通，最后再写一个模拟数据生成器。嗯，这三件事做完了，你的数据管道就有了“活水”。

2.1 Kafka 核心概念速览

先别急着装软件，花两分钟把几个关键概念理清楚。我在项目中见过不少同学把 Topic 和 Partition 搞混，结果数据写进去读不出来，排查半天才发现是分区策略配错了。

概念	一句话解释	我的经验
Topic	消息的逻辑分类，类似数据库里的表	命名规范很重要，我习惯用 `业务域.数据层.事件名`
Partition	Topic 的分片，每个 Partition 内部有序	分区数不是越多越好，跟消费并行度匹配就行
Offset	消息在 Partition 内的唯一序号	消费者重启后靠它续传，千万别手动乱改
Consumer Group	一组消费者共同消费一个 Topic	同一个 Group 里的消费者不会重复消费同一条消息
Broker	Kafka 服务器节点	生产环境至少 3 个 Broker，否则挂了就丢数据

核心原则：Kafka 不保证 Topic 级别的全局有序，只保证单个 Partition 内的消息有序。如果你的业务要求全局严格有序，要么用单个 Partition，要么在应用层自己排序。

2.2 安装与配置：从单机到集群

我建议你本地先跑一个单机版，把 API 调通了再上集群。下面是我在 Mac 和 Linux 上反复验证过的步骤，Windows 用户记得把路径里的斜杠改一下。

2.2.1 下载与解压

# 下载 Kafka（内置了 Zookeeper，开发环境够用）
wget https://downloads.apache.org/kafka/3.6.0/kafka_2.13-3.6.0.tgz
tar -xzf kafka_2.13-3.6.0.tgz
cd kafka_2.13-3.6.0

2.2.2 启动 Zookeeper 和 Kafka

# 启动 Zookeeper（Kafka 依赖它做协调）
bin/zookeeper-server-start.sh config/zookeeper.properties &

# 等几秒，再启动 Kafka Broker
bin/kafka-server-start.sh config/server.properties &

小技巧：启动后可以用 jps 命令检查进程。看到 QuorumPeerMain 和 Kafka 两个进程就说明跑起来了。我曾经有一次 Zookeeper 没启动就急着起 Kafka，结果 Broker 一直报连接拒绝，排查了十分钟才发现是启动顺序反了。

2.2.3 创建测试 Topic

# 创建一个名为 "test-topic" 的 Topic，3 个分区，1 个副本
bin/kafka-topics.sh --create \
  --bootstrap-server localhost:9092 \
  --replication-factor 1 \
  --partitions 3 \
  --topic test-topic

# 查看 Topic 列表
bin/kafka-topics.sh --list --bootstrap-server localhost:9092

2.2.4 生产环境配置要点

单机版能跑通就行，但真要上生产，有几个参数我建议你重点关注。我在一次大促压测时吃过亏——默认的 log.retention.hours 是 168 小时，结果磁盘被历史数据撑爆了，凌晨三点爬起来扩盘，那叫一个酸爽。

配置项	默认值	生产建议	原因
`log.retention.hours`	168	按业务需求设，比如 72	避免磁盘写满
`num.partitions`	1	3-6（根据吞吐量）	提高并行消费能力
`default.replication.factor`	1	3	保证高可用，挂一个 Broker 不丢数据
`min.insync.replicas`	1	2	配合 acks=all 防止数据丢失

2.3 生产者 API：把数据写进去

Kafka 的生产者 API 其实不复杂，核心就三步：配参数、建 Producer、发消息。但细节里全是坑，我一个个说。

2.3.1 基础生产者示例

import org.apache.kafka.clients.producer.*;

import java.util.Properties;

public class SimpleProducer {
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put("bootstrap.servers", "localhost:9092");
        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        // 重点：acks 设置
        props.put("acks", "all");
        props.put("retries", 3);
        props.put("batch.size", 16384);
        props.put("linger.ms", 5);

        KafkaProducer<String, String> producer = new KafkaProducer<>(props);

        for (int i = 0; i < 100; i++) {
            ProducerRecord<String, String> record =
                new ProducerRecord<>("test-topic", "key-" + i, "value-" + i);
            producer.send(record, (metadata, exception) -> {
                if (exception != null) {
                    System.err.println("发送失败: " + exception.getMessage());
                } else {
                    System.out.printf("发送成功: topic=%s, partition=%d, offset=%d%n",
                        metadata.topic(), metadata.partition(), metadata.offset());
                }
            });
        }
        producer.close();
    }
}

注意：acks=all 配合 min.insync.replicas=2 是最稳妥的配置。如果你追求极致吞吐量，可以设 acks=1，但极端情况下可能会丢一条消息。我个人的原则是：数据管道里跑的数据，宁可慢一点，不能丢一条。

2.3.2 分区策略

为什么要关心分区？因为分区决定了数据落到哪个 Broker，也决定了消费者怎么并行读。默认的分区器是按 key 的 hash 来分的，如果 key 是 null，就轮询。

我曾经遇到过一个场景：业务方要求同一个用户的所有订单数据必须落到同一个分区，这样才能保证消费端按用户维度做聚合时不出错。这时候就得自定义分区器。

// 自定义分区器：按用户 ID 取模
public class UserIdPartitioner implements Partitioner {
    @Override
    public int partition(String topic, Object key, byte[] keyBytes,
                         Object value, byte[] valueBytes, Cluster cluster) {
        int numPartitions = cluster.partitionCountForTopic(topic);
        String userId = (String) key;
        return Math.abs(userId.hashCode()) % numPartitions;
    }

    @Override
    public void close() {}

    @Override
    public void configure(Map<String, ?> configs) {}
}

2.4 消费者 API：把数据读出来

消费者比生产者稍微复杂一点，核心是理解 Consumer Group 和 Offset 提交。说白了，就是一群人怎么分工、怎么记录进度。

2.4.1 基础消费者示例

import org.apache.kafka.clients.consumer.*;
import java.time.Duration;
import java.util.*;

public class SimpleConsumer {
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put("bootstrap.servers", "localhost:9092");
        props.put("group.id", "test-group");
        props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        // 从最早的消息开始消费
        props.put("auto.offset.reset", "earliest");
        // 手动提交 offset
        props.put("enable.auto.commit", "false");

        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
        consumer.subscribe(Arrays.asList("test-topic"));

        try {
            while (true) {
                ConsumerRecords<String, String> records =
                    consumer.poll(Duration.ofMillis(100));
                for (ConsumerRecord<String, String> record : records) {
                    System.out.printf("partition=%d, offset=%d, key=%s, value=%s%n",
                        record.partition(), record.offset(), record.key(), record.value());
                }
                // 处理完后手动提交
                consumer.commitSync();
            }
        } finally {
            consumer.close();
        }
    }
}

为什么我坚持手动提交？自动提交虽然省事，但如果你在处理消息的过程中程序挂了，重启后会从上次自动提交的 offset 开始消费，中间那批没处理完的消息就丢了。手动提交可以让你在“确保处理完成”之后再提交，虽然代码多两行，但心里踏实。

2.4.2 消费者配置对比

配置项	选项	适用场景
`enable.auto.commit`	true / false	允许少量重复用 true，要求精确一次用 false
`auto.offset.reset`	earliest / latest / none	新消费者从头读用 earliest，只读新消息用 latest
`max.poll.records`	默认 500	处理慢的业务可以调小，防止超时 rebalance
`session.timeout.ms`	默认 45000	网络不稳定时可以适当调大

2.5 模拟数据生成器开发

环境搭好了，API 也调通了，但总不能用 Hello World 来测试管道吧？我习惯在项目初期就写一个模拟数据生成器，模拟真实业务的数据格式和流量。这样后面开发流处理逻辑的时候，直接用模拟数据就能验证。

2.5.1 设计思路

模拟数据生成器要干三件事：

生成结构化数据：比如订单、用户行为日志、传感器读数
控制发送速率：模拟高峰和低峰流量
写入 Kafka：用生产者 API 发到指定 Topic

2.5.2 代码实现

import org.apache.kafka.clients.producer.*;
import java.util.*;

public class MockDataGenerator {
    private static final String TOPIC = "order-events";
    private static final Random RANDOM = new Random();
    private static final String[] CITIES = {"北京", "上海", "广州", "深圳", "杭州"};
    private static final String[] PAY_TYPES = {"", "支付宝", "银行卡"};

    public static void main(String[] args) throws InterruptedException {
        Properties props = new Properties();
        props.put("bootstrap.servers", "localhost:9092");
        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        props.put("acks", "all");

        KafkaProducer<String, String> producer = new KafkaProducer<>(props);

        // 模拟持续生成数据，每秒 10 条
        while (true) {
            for (int i = 0; i < 10; i++) {
                String orderId = UUID.randomUUID().toString();
                String userId = "user_" + RANDOM.nextInt(1000);
                String city = CITIES[RANDOM.nextInt(CITIES.length)];
                String payType = PAY_TYPES[RANDOM.nextInt(PAY_TYPES.length)];
                double amount = 10.0 + RANDOM.nextDouble() * 990.0;

                // 构造 JSON 格式的消息
                String value = String.format(
                    "{\"orderId\":\"%s\",\"userId\":\"%s\",\"city\":\"%s\",\"payType\":\"%s\",\"amount\":%.2f}",
                    orderId, userId, city, payType, amount
                );

                ProducerRecord<String, String> record =
                    new ProducerRecord<>(TOPIC, userId, value);
                producer.send(record, (metadata, ex) -> {
                    if (ex != null) System.err.println("发送失败: " + ex.getMessage());
                });
            }
            // 每秒一批
            Thread.sleep(1000);
        }
    }
}

关键点：用 userId 作为消息的 key，这样同一个用户的所有订单会落到同一个分区。后面做窗口聚合时，这个设计能省掉很多麻烦。

2.6 本章知识体系总览

下面这张图把本章的核心脉络串起来了。从数据源到 Kafka，再到消费者，每一步都有对应的组件和配置。你可以把它贴在工位旁边，写代码时扫一眼就知道当前在哪个环节。

2.7 避坑指南

最后分享几个我踩过的坑，希望能帮你省点时间。

生产者超时重试导致消息重复：我曾经把 retries 设得很大，结果网络抖动时 Kafka 收到了重复消息。后来在消费端加了幂等性校验，用订单 ID 做去重。
消费者 Rebalance 频繁触发：有一次我开了 10 个消费者线程，但 Topic 只有 3 个分区，结果 7 个线程一直闲置，还频繁触发 Rebalance。记住：消费者线程数不要超过分区数。
Offset 提交失败导致重复消费：手动提交时如果业务处理时间超过 max.poll.interval.ms，消费者会被踢出 Group。解决方案是调大这个参数，或者把处理逻辑异步化。

好了，Kafka 的基础接入就讲到这里。环境搭好了，API 调通了，模拟数据也能跑了——接下来就可以开始真正的流批一体处理了。