2025年中国移动鸿鹄大数据实训营（大数据方向）kafka讲解及实践-第2次作业指导

书接上回，第二次作业比较容易解决，我问了ai，让他对我进行指导，按照它提供的步骤，我完成了本次实验，接下来我会标注出需要注意的细节，指导大家完成此次任务。

🎯 一、作业目标

✔️ 目标关键词：

✅ 搭建 Kafka 实现环境
✅ 使用 Java 写一个“网站订购数据实时统计程序”
✅ 保证数据可以：实时采集 → 实时消费 → 实时统计
✅ 提供完整演示文档、录屏资料

🚩 二、从头到尾的执行步骤 & 细节

✅准备阶段

要点：

第一步：Kafka 官方下载二进制版本

Apache Kafka我下的是3.9.1版本

第二步：JDK 必须安装（推荐 Oracle JDK 8 或 11，版本要和 IDEA 保持一致）我用的是8

要用邮箱注册一下Oracle账号，才能下载，不用虚拟机的话用windows版本的installer，下载后要记得目标文件的地址，下一步要用。

Java Downloads | Oracle 中国

第三步：配好 JAVA_HOME 和 PATH，必须 java -version 有输出

Win+r后cmd，检查你java成没成，

之后你电脑设置，高级系统设置，配置环境变量

✅ 配置环境变量（最重要）

1️⃣ 找到 系统环境变量 → Path
2️⃣ 把这个 JDK 的 bin 目录加进去，比如：

C:\Program Files\Java\jdk-11.0.20\bin

3️⃣ 新建 JAVA_HOME（可选但最好有）：

变量名：JAVA_HOME 变量值：C:\Program Files\Java\jdk-11.0.20

✅启动服务端

第四步：ZooKeeper：

先启动 ZooKeeper，新开一个窗口cmd：
cd /d D:\kafka
.\bin\windows\zookeeper-server-start.bat .\config\zookeeper.properties
正常输出：看到 binding to port 2181，就 OK。

别关，放后台即可

第五步：Kafka Broker：

新开一个窗口cmd，再启动 Kafka：
cd /d D:\kafka
.\bin\windows\kafka-server-start.bat .\config\server.properties
正常输出：看到 KafkaServer id=0 started，就 OK。

注意：

ZooKeeper 和 Kafka 都必须一直开着！
启动先后顺序是：ZooKeeper → Kafka

✅ 第六步：创建 Topic

再新开一个 cmd：
cd /d D:\kafka
.\bin\windows\kafka-topics.bat --create --topic order-topic --bootstrap-server localhost:9092 --partitions 1 --replication-factor 1

看到提示 Created topic order-topic. → 成功！

✅ 测试：命令行生产者 & 消费者（可选但很重要）

Producer：
.\bin\windows\kafka-console-producer.bat --topic order-topic --bootstrap-server localhost:9092

光标闪烁时输入：

hello kafka

Consumer：
.\bin\windows\kafka-console-consumer.bat --topic order-topic --bootstrap-server localhost:9092 --from-beginning

Consumer 窗口要能显示 Producer 发送的消息，证明消息管道通了。

✅ 用命令行测试 Producer / Consumer（可选但很有用）

这样能保证：

Kafka 能正常收发消息
防止你写 Java 程序时，结果没出来却不知道是 Kafka 还是代码问题

📌 1️⃣ 打开命令行 Producer

还是同样流程：

cd /d D:\kafka

输入：

.\bin\windows\kafka-console-producer.bat --topic order-topic --bootstrap-server localhost:9092

这时光标会闪烁，表示你可以直接输入内容了。

📌 2️⃣ 打开命令行 Consumer

再开一个新的 cmd：

cd /d D:\kafka

输入：

.\bin\windows\kafka-console-consumer.bat --topic order-topic --bootstrap-server localhost:9092 --from-beginning

📌 3️⃣ 测试

在 Producer 窗口输入：

hello

然后回车
在 Consumer 窗口应该立刻显示：

hello

✅ 如果收发正常，Kafka 配置一切正常！

✅ 第七步：iDEA 项目准备

创建 Maven 项目（方便自动管理 Kafka 依赖）
pom.xml：
<dependencies>
<dependency>
<groupId>org.apache.kafka</groupId>
<artifactId>kafka-clients</artifactId>
<version>3.7.0</version>
</dependency>
</dependencies>

⚠️ dependencies 必须在 <project> 里，别写到外面。3.7.0是kafka版本号，你下载什么版本就用什么号码

如果下载依赖慢，就加：
<repositories>
<repository>
<id>aliyun</id>
<url>https://maven.aliyun.com/repository/public</url>
</repository>
</repositories>
一定要点击 Reload Maven，看 External Libraries 有 kafka-clients。

以下是我的pom文件内容：

<project xmlns="http://maven.apache.org/POM/4.0.0"xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd"><modelVersion>4.0.0</modelVersion><groupId>org.example</groupId><artifactId>JavaProducerConsumer</artifactId><version>1.0-SNAPSHOT</version><properties><maven.compiler.source>1.8</maven.compiler.source><maven.compiler.target>1.8</maven.compiler.target></properties><dependencies><dependency><groupId>org.apache.kafka</groupId><artifactId>kafka-clients</artifactId><version>3.9.1</version></dependency></dependencies></project>

依赖写进去了之后，保存你的文件，要等它下载好，之后就是代码了

✅ 第八步：编写 Producer

在 src/main/java 下新建类 OrderProducer
Producer 要点：
- Kafka 地址：localhost:9092
- Topic 名：order-topic
- 代码核心：随机生成订单数据，循环发送
右键 Run，看到：
Sent: Order_123,456.78

就说明 Producer 已经把消息丢到 Kafka 了。

package com.example.kafka;import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.serialization.StringSerializer;import java.util.Properties;
import java.util.Random;public class OrderProducer {public static void main(String[] args) throws InterruptedException {Properties props = new Properties();props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());KafkaProducer<String, String> producer = new KafkaProducer<>(props);Random random = new Random();while (true) {String orderId = "Order_" + random.nextInt(1000);double amount = 100 + random.nextDouble() * 900;String msg = orderId + "," + amount;producer.send(new ProducerRecord<>("order-topic", orderId, msg));System.out.println("Sent: " + msg);Thread.sleep(1000); // 每秒一单}}
}

✅ 第九步：编写 Consumer

新建 OrderConsumer
Consumer 要点：
- Kafka 地址：localhost:9092
- Topic 名：order-topic
- 逻辑：循环消费消息、解析金额、累计统计、打印实时结果
右键 Run，会实时输出：
Received: Order_123,456.78
Total Orders: X, Total Amount: Y

package com.example.kafka;import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.serialization.StringDeserializer;import java.time.Duration;
import java.util.Collections;
import java.util.Properties;public class OrderConsumer {public static void main(String[] args) {Properties props = new Properties();props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());props.put(ConsumerConfig.GROUP_ID_CONFIG, "order-consumer-group");KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);consumer.subscribe(Collections.singletonList("order-topic"));int totalOrders = 0;double totalAmount = 0;while (true) {ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000));for (ConsumerRecord<String, String> record : records) {String[] parts = record.value().split(",");double amount = Double.parseDouble(parts[1]);totalOrders++;totalAmount += amount;System.out.println("Received: " + record.value());System.out.println("Total Orders: " + totalOrders + ", Total Amount: " + totalAmount);}}}
}

✅第十步：验证功能完整

先运行producer，再运行consumer，截图，或者录屏，（如果能运行代码，你知道怎么截图和录屏的）因为这就是你做成这个实验的成果

✔ Producer 持续发
✔ Consumer 实时收
✔ Consumer 实时汇总统计

这里录个屏，截几张图就行

✅ 结果整理（交付要点）

🌟 交什么？