书接上文 《Kafka 从入门到实践》——Kafka 负责消息去哪儿Flink 负责数据怎么变。读完这篇你将能够用 Flink 消费 Kafka 数据写自己的处理算子把结果写回 Kafka。一、Flink 是什么为什么需要它1.1 一个场景上篇博客我们搭好了 Kafka上传 .dat 日志文件 → file-topic → 消费者保存到本地。现在老板说别光存啊给我分析每行有多少字符、多少单词把结果存到另一个 topic。你当然可以在 Spring 消费者里加一段分析代码KafkaListener(topics file-topic, ...) public void onFile(ConsumerRecordString, byte[] record) { String content new String(record.value()); for (String line : content.split(\n)) { // 分析、统计、再发到 processed-topic... } }能跑。但问题是文件大了怎么办100MB 的文件在一个消费者线程里逐行处理内存直接爆了想加个每分钟统计日志级别数量的需求代码越来越臃肿三个消费者都要做同样的分析代码复制三份你需要一个专门干数据处理的框架——这就是 Flink。1.2 Flink vs 自己写代码自己在 Consumer 里写用 Flink大数据量单线程扛不住得自己写多线程自动分布式并行处理可靠性崩溃后手动补数据自动 checkpoint重启后精确恢复处理逻辑跟消费逻辑耦合在一起Source / Transform / Sink 清晰分离扩展性加机器要改代码加个 TaskManager 就完事复杂计算自己实现窗口、聚合、join内置 window、aggregate、connect一句话Kafka 负责可靠地传递数据Flink 负责对数据做实时计算。二、核心概念用人话解释2.1 一张图看懂 Flink 的三板斧Source数据入口 Transform转换 Sink数据出口 │ │ │ ▼ ▼ ▼ ┌──────────┐ byte[] ┌──────────┐ String ┌──────────┐ │ Kafka │──────────▶│ flatMap │──────────▶│ Kafka │ │ Source │ │ 逐行解析 │ │ Sink │ └──────────┘ └──────────┘ └──────────┘每个环节在 Flink 里叫一个算子Operator算子类型干什么的类比Source从外部系统拉数据水管工把水从水库引入管道Transform对数据做计算、转换净水器水进去→过滤→出来Sink把处理结果写回外部系统水管工把水送到你家水龙头2.2 几个关键概念概念一句话解释类比DataStream无限的数据流一条永不关闸的水管数据就是水流flatMap一个输入零到多个输出拆包裹一个大箱子进去里面一个个小盒子出来map一个输入恰好一个输出翻译机一句中文进一句英文出filter通过条件才放行筛子大颗粒留下小颗粒穿过并行度同时干活的人数3 个人同时拧 3 个水龙头Checkpoint处理进度的快照游戏存档崩溃后读档继续Watermark事件时间的标记迟到容忍度——超过这个时间的数据就不等了2.3 算子的序列化要求踩坑点Flink 会把你的算子对象序列化成字节发送到不同机器上执行。所以❌ 匿名内部类.flatMap((s, out) - { ... })← Flink 序列化不了 lambda❌ 非静态内部类隐式持有外部类的this引用外部类可能不可序列化✅public static class没有外部依赖可以安全序列化// ✅ 正确写法必须 public static public static class DatLineProcessor implements FlatMapFunctionbyte[], String { Override public void flatMap(byte[] value, CollectorString out) { // 你的处理逻辑 } }三、环境准备3.1 依赖properties flink.version1.19.1/flink.version !-- 用 1.19别用 1.18有坑 -- /properties dependencies !-- Flink 流处理核心 -- dependency groupIdorg.apache.flink/groupId artifactIdflink-streaming-java/artifactId version${flink.version}/version !-- 排除 log4j-slf4j-impl避免和 Spring Boot 的 Logback 抢 SLF4J 绑定 -- exclusions exclusion groupIdorg.apache.logging.log4j/groupId artifactIdlog4j-slf4j-impl/artifactId /exclusion /exclusions /dependency !-- Flink Kafka 连接器 -- dependency groupIdorg.apache.flink/groupId artifactIdflink-connector-kafka/artifactId version3.2.0-1.19/version !-- 版本号必须跟 Flink 版本匹配 -- /dependency !-- KafkaSource 需要这个基础库 -- dependency groupIdorg.apache.flink/groupId artifactIdflink-connector-base/artifactId version${flink.version}/version /dependency !-- 本地运行客户端 -- dependency groupIdorg.apache.flink/groupId artifactIdflink-clients/artifactId version${flink.version}/version /dependency /dependencies版本兼容性是个大坑详见第七章。3.2 前提条件确保你已经按上篇博客搭好了 Kafka 集群4 节点 KRaft并且依赖里已经有 Spring Kafka。四、实战构建 Kafka → Flink → Kafka 管线4.1 完整架构┌──────────┐ ┌──────────┐ ┌──────────┐ ┌──────────┐ │ 用户上传 │ │ Kafka │ │ Flink │ │ Kafka │ │ .dat 文件 │────▶│file-topic│────▶│ 流处理 │────▶│processed │ └──────────┘ │ (byte[]) │ │ 逐行分析 │ │ -topic │ └──────────┘ └──────────┘ │(String) │ └────┬─────┘ │ ┌─────────────▼─────────────┐ │ Spring Kafka 消费者 │ │ 把分析结果存到内存API 返回 │ └───────────────────────────┘4.2 第一步写 Flink Jobpublic class DataFileProcessingJob implements Serializable { private final String bootstrapServers; public DataFileProcessingJob(String bootstrapServers) { this.bootstrapServers bootstrapServers; } public void start() throws Exception { // 1. 创建本地执行环境开发和测试用这一个就够了 Configuration flinkConfig new Configuration(); flinkConfig.set(RestOptions.BIND_PORT, 0); // 随机端口避免冲突 final StreamExecutionEnvironment env StreamExecutionEnvironment.createLocalEnvironment(1, flinkConfig); // 2. Kafka Source —— 从 file-topic 拉取 byte[] 消息 KafkaSourcebyte[] source KafkaSource.byte[]builder() .setBootstrapServers(bootstrapServers) .setTopics(file-topic) .setGroupId(flink-file-processor) .setStartingOffsets(OffsetsInitializer.latest()) .setValueOnlyDeserializer(new RawBytesDeserializer()) .build(); DataStreambyte[] rawStream env.fromSource( source, WatermarkStrategy.noWatermarks(), Kafka-Source); // 3. Transform —— flatMap 拆成逐行 DataStreamString processed rawStream .flatMap(new DatLineProcessor()) .name(Parse-Lines); // 4. Kafka Sink —— 分析结果写回 processed-topic KafkaSinkString sink KafkaSink.Stringbuilder() .setBootstrapServers(bootstrapServers) .setRecordSerializer( KafkaRecordSerializationSchema.builder() .setTopic(processed-topic) .setValueSerializationSchema(new SimpleStringSchema()) .build()) .build(); processed.sinkTo(sink); // 5. 启动这个方法会阻塞直到任务被取消 env.execute(Flink-DAT-Processor); } // // 算子类必须是 public static // /** 字节数组反序列化器 */ public static class RawBytesDeserializer extends AbstractDeserializationSchemabyte[] { Override public byte[] deserialize(byte[] message) { return message; } } /** 逐行处理器 */ public static class DatLineProcessor implements FlatMapFunctionbyte[], String { Override public void flatMap(byte[] fileData, CollectorString out) { String content new String(fileData, StandardCharsets.UTF_8); int lineNumber 0; for (String line : content.split(\\r?\\n)) { lineNumber; if (line.trim().isEmpty()) continue; int chars line.length(); int words line.trim().split(\\s).length; String summary line.length() 50 ? line.substring(0, 50) ... : line; out.collect(String.format( L%d | 字:%d | 词:%d | %s, lineNumber, chars, words, summary)); } } } }4.3 第二步在 Spring Boot 中管理 Flink 生命周期env.execute()是阻塞调用——它会一直占据当前线程直到任务被取消。如果在 Spring Boot 主线程里直接调用整个应用就卡死了。所以必须放在独立线程里运行Service public class FlinkJobService { Value(${spring.kafka.bootstrap-servers}) private String bootstrapServers; private final AtomicBoolean running new AtomicBoolean(false); private ExecutorService executor; /** 在独立线程中启动 Flink Job */ public synchronized void startJob() { if (running.get()) return; running.set(true); executor Executors.newSingleThreadExecutor(r - { Thread t new Thread(r, flink-job); t.setDaemon(true); // 守护线程主线程退出时自动结束 return t; }); executor.submit(() - { try { new DataFileProcessingJob(bootstrapServers).start(); } catch (Exception e) { e.printStackTrace(); } finally { running.set(false); } }); } /** 停止 Flink Job */ public synchronized void stopJob() { if (!running.get()) return; running.set(false); if (executor ! null) executor.shutdownNow(); } public boolean isRunning() { return running.get(); } PreDestroy public void onShutdown() { stopJob(); } }关键点t.setDaemon(true)保证 Spring Boot 关闭时 Flink 线程不会阻止 JVM 退出shutdownNow()会中断env.execute()的阻塞让任务停止4.4 第三步消费处理结果Flink 把分析结果写回了processed-topic再用 Spring Kafka 监听它Service public class FlinkResultConsumer { private final ListString results Collections.synchronizedList(new ArrayList()); KafkaListener(topics processed-topic, groupId flink-result-group) public void onResult(String result) { System.out.println([Flink结果] result); results.add(result); } public ListString getResults() { return new ArrayList(results); } }4.5 第四步加个 Controller用 REST 控制RestController RequestMapping(/api/flink) public class FlinkController { private final FlinkJobService flinkJobService; private final FlinkResultConsumer resultConsumer; PostMapping(/start) public MapString, Object start() { flinkJobService.startJob(); return Map.of(status, ok, running, flinkJobService.isRunning()); } PostMapping(/stop) public MapString, Object stop() { flinkJobService.stopJob(); return Map.of(status, ok, running, flinkJobService.isRunning()); } GetMapping(/status) public MapString, Object status() { return Map.of(running, flinkJobService.isRunning()); } GetMapping(/results) public MapString, Object results() { ListString list resultConsumer.getResults(); return Map.of(count, list.size(), results, list); } DeleteMapping(/results) public MapString, Object clear() { resultConsumer.clearResults(); return Map.of(status, ok); } }4.6 跑起来# 1. 启动应用 mvn spring-boot:run # 2. 启动 Flink Job curl -X POST http://localhost:8080/api/flink/start # 3. 上传一个 .dat 文件 curl -X POST http://localhost:8080/api/file/upload -F filesample.dat # 4. 查看 Flink 分析结果 curl http://localhost:8080/api/flink/results结果长这样{ count: 7, results: [ L1 | 字:57 | 词:6 | 2024-01-15 08:30:22 INFO 用户登录 userId1001 ip192.1..., L2 | 字:76 | 词:7 | 2024-01-15 08:31:05 INFO 查询订单 orderIdORD-2024-001..., L3 | 字:76 | 词:7 | 2024-01-15 08:32:18 INFO 创建订单 orderIdORD-2024-002..., L4 | 字:91 | 词:8 | 2024-01-15 08:33:47 WARN 库存不足 productIdPROD-887 s..., L5 | 字:81 | 词:7 | 2024-01-15 08:34:12 INFO 支付成功 orderIdORD-2024-001..., L6 | 字:75 | 词:7 | 2024-01-15 08:35:03 ERROR 支付失败 orderIdORD-2024-00..., L7 | 字:58 | 词:6 | 2024-01-15 08:36:21 INFO 用户登出 userId1001 sessionT... ] }7 行日志每行都统计了字符数、单词数超过 50 字符自动截断。这就是一条完整的 Kafka → Flink → Kafka 流处理管线。五、Flink 常用算子速查5.1 Transform 算子算子输入→输出用途示例map1→1逐条转换dataStream.map(s - s.toUpperCase())flatMap1→N一条拆成多条dataStream.flatMap(new LineSplitter())filter1→0或1过滤dataStream.filter(s - s.contains(ERROR))keyBy分组按 key 分流dataStream.keyBy(Log::getLevel)reduce聚合合并计算stream.keyBy(...).reduce((a, b) - a b)window时间窗口按时间段聚合stream.keyBy(...).window(TumblingEventTimeWindows.of(Time.seconds(5)))5.2 Source / Sink类型常见的SourceKafka、文件、Socket、JDBC、自定义SinkKafka、文件、JDBC、Redis、Elasticsearch、自定义5.3 如何选择简单转换一行变一行 → map 一行拆成多行 → flatMap 只要某些数据 → filter 按某个字段分组统计 → keyBy reduce/window 多流合并 → connect / union六、Flink 在 Spring Boot 中的架构细节6.1 为什么要在独立线程运行Spring Boot 主线程 Flink 守护线程 │ │ │ startJob() ───────────────────────────▶│ env.execute() ← 阻塞在这里 │ (立即返回) │ │ │ │ │ 轮询 Kafka处理数据 │ 处理 HTTP 请求 │ │ │ 响应 /api/flink/status │ │ │ │ │ │ stopJob() ─── shutdownNow() ───────────▶│ InterruptedException │ │ ▼ ▼ 正常运行 安全退出6.2 关键配置项// 并行度 1开发环境一个 TaskManager 就够了 StreamExecutionEnvironment.createLocalEnvironment(1, flinkConfig); // 随机 REST 端口避免多实例冲突 flinkConfig.set(RestOptions.BIND_PORT, 0); // 起始消费位置 OffsetsInitializer.earliest() // 从最早的消息开始适合开发测试会重复消费 OffsetsInitializer.latest() // 只消费新消息适合生产环境七、踩坑记录坑1flink-connector-kafka 版本不兼容现象Flink 启动后立刻停止没有任何报错也没有消费到任何数据。原因flink-connector-kafka:3.0.2-1.18 ← 针对 kafka-clients 3.4.0 编译 Spring Boot 3.3.0 引入的 kafka-clients ← 实际 3.7.0二进制不兼容导致 Flink 内部 Kafka 消费者静默失败。解决升级到兼容版本组合。组件❌ 旧版本✅ 新版本Flink1.18.11.19.1flink-connector-kafka3.0.2-1.183.2.0-1.19flink-connector-base无1.19.1新增版本选择黄金法则Flink 版本 flink-connector-kafka 后缀版本如3.2.0-1.19→ Flink 1.19flink-connector-kafka 的 kafka-clients 版本 ≥ Spring Kafka 的 kafka-clients 版本坑2Logback 和 Log4j 打架Flink 自带 Log4jSpring Boot 用 Logback。两个日志框架抢 SLF4J 绑定。解决只排除log4j-slf4j-impl保留log4j-api和log4j-core。exclusion groupIdorg.apache.logging.log4j/groupId artifactIdlog4j-slf4j-impl/artifactId /exclusion !-- 不要排除 log4j-api 和 log4j-coreFlink 需要它们 --坑3算子类序列化失败现象NotSerializableException。原因用了非静态内部类或 lambda。// ❌ 这样写 Flink 会报 NotSerializableException DataStreamString processed rawStream.flatMap( (byte[] data, CollectorString out) - { // lambda 不能被 Flink 序列化 } ); // ✅ 必须单独定义一个 public static class public static class DatLineProcessor implements FlatMapFunctionbyte[], String { Override public void flatMap(byte[] value, CollectorString out) { ... } }坑4env.execute() 阻塞导致 Spring Boot 假死现象调用任何 API 都无响应。原因在主线程里直接调了env.execute()。解决放独立线程设setDaemon(true)。八、常见问题 FAQQ1: Flink 和 Spark Streaming 有什么区别FlinkSpark Streaming处理模型真正的逐条流处理微批次攒一小批再处理延迟毫秒级秒级状态管理内置 State Backend非常成熟相对弱一些学习曲线中等如果你已经会 Spark 就比较简单适用场景低延迟实时计算Lambda 架构、与 Spark 生态集成简单记真正的实时用 Flink准实时用 Spark Streaming。Q2: Flink 能处理多大的数据量阿里用 Flink 处理双十一的每秒几十亿条交易消息。你的 .dat 文件完全不用担心。Q3: 处理挂了怎么办会丢数据吗Flink 有Checkpoint 机制——定期把处理进度当前读到哪个 offset、中间计算结果存到持久化存储。挂了之后从最近的 checkpoint 恢复不会丢数据。本文用的是本地开发模式没开 checkpoint。生产环境加几行配置就开启了env.enableCheckpointing(60_000); // 每 60 秒做一次快照 env.getCheckpointConfig().setCheckpointStorage(file:///checkpoints);Q4: Flink Job 一定要在 Spring Boot 里跑吗不一定。Flink 的标准部署方式是提交到Flink ClusterJobManager 多个 TaskManager开发环境本地跑createLocalEnvironment 生产环境提交到 Flink 集群flink run -c MainClass xxx.jar本文为了演示方便嵌在了 Spring Boot 里。生产环境建议独立部署 Flink 集群。Q5: 怎么知道 Flink 有没有在处理数据看控制台输出。正常运行的日志长这样[Flink-Source] 收到消息, 大小: 585 bytes [Flink-Transform] 处理完成, 共 7 行 [Flink结果] L1 | 字:57 | 词:6 | 2024-01-15 08:30:22 INFO...也可以通过 Flink 的 Web UI默认localhost:8081查看详细的吞吐量、延迟指标。九、学习路线图第1步 ✅ 理解核心概念 ├── Source → Transform → Sink ├── DataStream / flatMap / map / filter └── 并行度 / Checkpoint 第2步 ✅ 本地跑通第一条管线 ├── KafkaSource → flatMap → KafkaSink ├── 理解序列化要求public static class └── Spring Boot 中管理 Flink 生命周期 第3步 ✅ 踩坑 调试 ├── 版本兼容性Flink Kafka connector kafka-clients ├── 日志冲突Log4j vs Logback └── 算子序列化问题 第4步进阶探索 ├── Window 窗口计算每分钟聚合、滑动窗口 ├── KeyBy Reduce 分组聚合 ├── Checkpoint Savepoint故障恢复 ├── Event Time vs Processing Time事件时间处理 ├── CEP复杂事件匹配如30秒内连续3次登录失败 └── 提交到 Flink 集群JobManager / TaskManager十、总结回顾一下你在这篇博客里掌握了什么Flink 的定位不是替代 Kafka而是和 Kafka 配合——Kafka 负责传数据Flink 负责算数据三板斧模式Source 拉数据 → Transform 处理 → Sink 写结果所有 Flink Job 都是这个套路算子必须 public static因为 Flink 要序列化它们发到各台机器执行Spring Boot 集成env.execute()会阻塞必须在独立守护线程里运行版本兼容Flink 1.19.1 flink-connector-kafka 3.2.0-1.19 kafka-clients 3.7.0 是一组可用的版本组合从 Kafka 到 Flink你已经掌握了一条完整的实时数据处理链路消息怎么传Kafka→ 数据怎么算Flink→ 结果怎么用Spring Boot API。剩下的就是在这个骨架上不断丰富业务逻辑。
