📚 课程目录第4章实时流处理引擎

4、实时流处理引擎：Apache Flink入门、Flink运行架构、Flink开发环境搭建、第一个Flink WordCount程序

好，咱们进入第四章。前面聊完了数据采集和消息队列，现在该轮到真正的「流处理引擎」上场了。

你想想看，数据从 Kafka 里源源不断地流过来，总得有个东西能实时地算一算、聚合一下、做个规则判断吧？Spark Streaming 当然也行，但我个人更偏爱 Flink。为什么？因为 Flink 是「真·流处理」，每条数据来了就处理，延迟在毫秒级。而 Spark Streaming 本质上是微批次，攒一批再算，延迟至少几秒。

我记得 2019 年在一个电商大促项目里，业务方要求「用户点击商品后 500 毫秒内必须看到推荐结果变化」。用 Spark Streaming 死活压不到这个指标，换成 Flink 后，300 毫秒就搞定了。从那以后，我对 Flink 就特别有好感。

4.1 为什么是 Flink？

先简单说说 Flink 的核心优势：

真正的流处理：每条数据独立处理，不是微批次。
Exactly-Once 语义：数据不丢不重，做实时推荐时这点特别重要——你肯定不想给用户重复推荐同一个商品吧？
高吞吐、低延迟：单节点每秒能处理百万级事件，延迟却只有几十毫秒。
状态管理：Flink 内置了强大的状态后端，可以记住过去的数据状态，做窗口聚合、去重、模式匹配都很方便。

说白了，Flink 就是为「实时」而生的。如果你要做实时推荐、实时风控、实时大屏，选 Flink 基本不会错。

4.2 Flink 运行架构

理解 Flink 的架构，是后面写作业、调优的基础。我画了一张图，帮你把核心组件串起来。

Flink 运行时架构核心组件

简单解释一下这张图：

Client：你写的 Flink 程序，打包成 jar 包，通过 Client 提交到集群。
JobManager：集群的「大脑」。负责任务调度、Checkpoint 协调、故障恢复。如果 JobManager 挂了，整个集群就瘫痪了——所以生产环境一定要配高可用。
TaskManager：真正干活的「工人」。每个 TaskManager 有多个 Slot，Slot 就是执行任务的最小单元。一个 Slot 可以跑一个线程。
数据源与输出：Flink 支持各种 Source 和 Sink，比如 Kafka、文件、Socket、数据库等。

💡 我的经验：刚开始学 Flink 时，我总搞不清 Slot 和并行度的关系。后来我总结了一句话：并行度决定你能同时跑多少个任务，Slot 数量决定你能同时跑多少个任务实例。比如并行度设为 4，但只有 2 个 Slot，那对不起，剩下的 2 个任务只能排队等着。

4.3 Flink 开发环境搭建

好，理论说完了，咱们动手搭环境。我习惯在本地用 IDEA 开发，然后提交到集群测试。下面是我的标准步骤：

4.3.1 前置条件

JDK 8 或 11（推荐 JDK 8，兼容性最好）
Maven 3.6+
IDEA（社区版就够用）
Git（可选，但建议装）

4.3.2 创建 Maven 项目

在 IDEA 里新建一个 Maven 项目，然后在 pom.xml 里加入 Flink 依赖。我直接给你一个可用的配置：

<properties>
    <flink.version>1.17.0</flink.version>
    <scala.binary.version>2.12</scala.binary.version>
</properties>

<dependencies>
    <!-- Flink Java 核心依赖 -->
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-java</artifactId>
        <version>${flink.version}</version>
    </dependency>
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-streaming-java</artifactId>
        <version>${flink.version}</version>
        <scope>provided</scope>
    </dependency>
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-clients</artifactId>
        <version>${flink.version}</version>
    </dependency>
</dependencies>

⚠️ 注意：flink-streaming-java 的 scope 设为 provided，是因为 Flink 集群自带了这些 jar 包。如果你打成 fat jar 提交，可能会和集群的版本冲突。我曾经因为这个坑，浪费了整整一个下午排查 ClassNotFoundException。

4.3.3 配置 Log4j（可选但推荐）

在 src/main/resources 下新建 log4j.properties：

log4j.rootLogger=INFO, console
log4j.appender.console=org.apache.log4j.ConsoleAppender
log4j.appender.console.layout=org.apache.log4j.PatternLayout
log4j.appender.console.layout.ConversionPattern=%d{yyyy-MM-dd HH:mm:ss} %-5p %c{1}:%L - %m%n

这样你就能在控制台看到 Flink 的日志了，调试起来方便很多。

4.4 第一个 Flink WordCount 程序

环境搭好了，咱们写个经典的 WordCount。不过这次是「流式」的——数据源源不断地进来，Flink 实时统计单词出现次数。

4.4.1 代码实现

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

public class StreamingWordCount {
    public static void main(String[] args) throws Exception {
        // 1. 创建流执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // 2. 从 socket 读取数据（本地测试用）
        // 启动方式：nc -lk 9999
        DataStream<String> text = env.socketTextStream("localhost", 9999);

        // 3. 数据转换：切分单词，计数
        DataStream<Tuple2<String, Integer>> counts = text
                .flatMap(new Tokenizer())
                .keyBy(value -> value.f0)   // 按单词分组
                .sum(1);                    // 累加次数

        // 4. 打印结果
        counts.print();

        // 5. 启动任务
        env.execute("Streaming WordCount");
    }

    // 自定义 FlatMapFunction：将每行文本切分成单词
    public static class Tokenizer implements FlatMapFunction<String, Tuple2<String, Integer>> {
        @Override
        public void flatMap(String value, Collector<Tuple2<String, Integer>> out) {
            // 按空格切分
            String[] words = value.toLowerCase().split("\\W+");
            for (String word : words) {
                if (word.length() > 0) {
                    out.collect(new Tuple2<>(word, 1));
                }
            }
        }
    }
}

4.4.2 运行测试

先在终端启动一个 socket 数据源：nc -lk 9999
在 IDEA 里直接运行 StreamingWordCount 的 main 方法
在终端输入一些英文单词，比如：hello world hello flink
观察 IDEA 控制台输出：

(hello,1)
(world,1)
(hello,2)
(flink,1)

看到了吗？Flink 实时地统计了每个单词出现的次数。你再输入一次 hello，它会输出 (hello,3)——这就是「有状态的计算」。

🔑 关键点：keyBy 之后，Flink 会把相同单词的数据路由到同一个 TaskManager 的同一个 Slot 里，然后 sum(1) 会维护一个状态，记录当前单词的累计次数。这就是 Flink 状态管理的雏形。

4.5 常见问题与避坑

写第一个 Flink 程序时，你可能会遇到这些问题。我当年都踩过：

问题	原因	解决方案
ClassNotFoundException	依赖 scope 设置错误，或 jar 包冲突	检查 pom.xml 中 flink-streaming-java 的 scope 是否为 provided
连接超时	socket 数据源未启动，或端口被占用	先运行 nc -lk 9999，确保端口可用
输出结果不更新	没有调用 env.execute()	Flink 是懒加载，必须显式调用 execute 才会启动任务
中文乱码	socket 默认编码问题	在 socketTextStream 中指定编码：`socketTextStream("localhost", 9999, "\n", Charset.forName("UTF-8"))`

💡 调试小技巧：如果你不想开 socket，也可以用 env.fromElements("hello world", "hello flink") 直接造几条测试数据。这样调试起来更快，不用每次都在终端敲单词。

4.6 本章小结

这一章我们干了三件事：

理解了 Flink 的运行架构——Client、JobManager、TaskManager 各司其职
搭好了 Flink 开发环境——Maven 项目 + 依赖配置
跑通了第一个流式 WordCount——从 socket 读数据，实时统计单词次数

说实话，WordCount 虽然简单，但它包含了 Flink 最核心的几个概念：Source、Transformation、Sink、KeyBy、State。后面的复杂应用，都是在这个基础上叠加窗口、时间语义、状态后端等高级特性。

嗯，这一章就到这里。代码建议你亲手敲一遍，别复制粘贴。敲完你会对 Flink 的 API 有更深的体感。