环境准备：Kafka 集群搭建与第一个 Streams 程序

说实话，很多初学者一上来就急着写代码，结果环境没配好，折腾半天跑不起来。我见过不少同事卡在这一步，白白浪费一整天。所以咱们先把地基打牢——这一章我会带你搞定三件事：搭一个能用的 Kafka 集群、把 Streams 依赖引进来、再跑通一个 Hello World。

Kafka 集群搭建：单机版就够了

学习阶段没必要搞三台机器。单机模式完全够用。我个人习惯用 伪分布式——就是一台机器上起多个 Broker 进程。这样既能体验集群特性，又省资源。

第一步：下载和解压

去 Kafka 官网下载二进制包。我建议用 2.8.0 以上版本，因为从 2.8 开始支持 KRaft 模式，不用再依赖 Zookeeper。不过为了兼容性，咱们还是用传统的 ZK 模式，毕竟大部分生产环境还在用。

# 下载（以 3.2.0 为例）
wget https://downloads.apache.org/kafka/3.2.0/kafka_2.13-3.2.0.tgz

# 解压
tar -xzf kafka_2.13-3.2.0.tgz
cd kafka_2.13-3.2.0

第二步：启动 Zookeeper

Kafka 依赖 ZK 做元数据管理。嗯，这里要注意——先启动 ZK，再启动 Kafka，顺序反了会报错。

# 启动 Zookeeper（默认端口 2181）
bin/zookeeper-server-start.sh config/zookeeper.properties

看到 binding to port 0.0.0.0/0.0.0.0:2181 就说明起来了。

第三步：启动 Kafka Broker

单机模式直接启动一个 Broker 就行：

# 启动 Kafka（默认端口 9092）
bin/kafka-server-start.sh config/server.properties

如果想玩伪分布式，复制两份配置文件，改三个地方：

配置项	Broker 1	Broker 2	说明
broker.id	0	1	集群内唯一标识
listeners	PLAINTEXT://localhost:9092	PLAINTEXT://localhost:9093	端口不能冲突
log.dirs	/tmp/kafka-logs-0	/tmp/kafka-logs-1	数据目录隔离

第四步：验证集群

创建一个测试 Topic，看看能不能正常读写：

# 创建 Topic
bin/kafka-topics.sh --create --topic test-topic \
  --bootstrap-server localhost:9092 \
  --partitions 3 --replication-factor 1

# 启动生产者，随便输入几条消息
bin/kafka-console-producer.sh --topic test-topic \
  --bootstrap-server localhost:9092

# 另开终端，启动消费者
bin/kafka-console-consumer.sh --topic test-topic \
  --bootstrap-server localhost:9092 \
  --from-beginning

生产者输入 hello kafka，消费者能收到，就说明集群没问题。

Kafka Streams 依赖引入

集群搭好了，接下来把 Streams 库加到项目里。我用 Maven 比较多，Gradle 也顺手写一份。

Maven 方式

<dependency>
    <groupId>org.apache.kafka</groupId>
    <artifactId>kafka-streams</artifactId>
    <version>3.2.0</version>
</dependency>

Gradle 方式

implementation 'org.apache.kafka:kafka-streams:3.2.0'

版本号要和你的 Kafka 集群保持一致。我见过有人用 2.8 的客户端连 3.0 的集群，虽然大部分功能兼容，但某些新特性会报错。

第一个 Hello World 程序

写一个最简单的 Streams 应用：从输入 Topic 读消息，原样写到输出 Topic。说白了就是一个透传。

代码实现

import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.streams.KafkaStreams;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.StreamsConfig;
import org.apache.kafka.streams.kstream.KStream;

import java.util.Properties;

public class HelloWorldStreams {

    public static void main(String[] args) {
        // 1. 配置
        Properties props = new Properties();
        props.put(StreamsConfig.APPLICATION_ID_CONFIG, "hello-world-app");
        props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());
        props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass());

        // 2. 构建拓扑
        StreamsBuilder builder = new StreamsBuilder();
        KStream<String, String> source = builder.stream("input-topic");
        source.to("output-topic");

        // 3. 启动
        KafkaStreams streams = new KafkaStreams(builder.build(), props);
        streams.start();

        // 优雅关闭
        Runtime.getRuntime().addShutdownHook(new Thread(streams::close));
    }
}

运行测试

先创建两个 Topic：

bin/kafka-topics.sh --create --topic input-topic \
  --bootstrap-server localhost:9092 --partitions 1 --replication-factor 1

bin/kafka-topics.sh --create --topic output-topic \
  --bootstrap-server localhost:9092 --partitions 1 --replication-factor 1

然后运行你的 main 方法。在生产者端往 input-topic 发消息：

bin/kafka-console-producer.sh --topic input-topic \
  --bootstrap-server localhost:9092
> Hello Kafka Streams!

另开终端消费 output-topic：

bin/kafka-console-consumer.sh --topic output-topic \
  --bootstrap-server localhost:9092 --from-beginning
> Hello Kafka Streams!

看到消息原样输出，恭喜你——第一个 Streams 程序跑通了！

本章知识体系

下面这张图把环境准备阶段的核心流程串起来了：