简介：这是一份面向大数据流处理开发者的Flink实战代码包，聚焦从Kafka消费实时数据、完成业务计算后分别写入Redis集群与MySQL的完整链路，可用于实时监控、日志分析和在线广告等低延迟场景。资源共145个文件，约48.47MB，以java源码、编译后的class、xml配置为主，另含properties配置、jar依赖及mvnw构建脚本，便于理解工程结构并直接运行调试。代码演示了事件日志解析、自定义水位线提取、keyBy分组与window窗口统计等关键环节，并给出Kafka连接器参数、Redis Sink连接设置以及JDBC批量写入MySQL的实现方式，覆盖从数据接入、计算到多端存储的典型流程。适合有一定Flink基础、希望上手真实项目的学习者。目前已有597人学习下载，通过研读源码与配置，可掌握流式数据落地的常见设计思路和排错要点。
1. 拿到一个叫“flink读取kafka数据”的压缩包，你真正需要的是什么
拿到一个叫 flink读取kafka数据 的压缩包，里面大概率是一份可直接运行的 Flink 工程，或者是一套配套教程的 demo 源码。这个标题指向的正是实时计算里最常被问到的链路：Kafka 里堆着千万级消息，Flink 把它们拉下来做处理，再交给你下游的存储或告警。它能解决的核心问题不是“能不能读”，而是怎么把这条链路从零跑通，并且跑得稳。适合刚接手实时任务、想验证 Flink 与 Kafka 连接方式的读者，也适合课上跑通了 demo 但没上过生产的同学——代码能起来是一回事，参数和坑是另一回事。这篇就按我平时搭这类工程的顺序，从环境选型、三种读取方式、调参基线一直说到排障清单。
2. 把底座备齐：Kafka 与 Flink 的版本选型和本地环境
2.1 版本选型：JDK、Kafka、Flink 之间怎么搭不翻车
排除掉业务逻辑之前，先排除版本问题。很多 demo 跑不起来不是代码错了，是 Flink 连接器坐标和 Kafka 客户端版本对不上。Flink 1.14 及以前，Kafka 连接器叫flink-connector-kafka_2.11或_2.12，版本号跟着 Flink 走；从 1.15 开始官方把连接器独立发布，坐标变成flink-connector-kafka，后面的版本号要单独匹配 Flink 主版本。我一般是 Flink 1.17.x 配 Kafka 3.x、JDK 11，这套组合踩坑最少。
| Flink 版本 | Kafka 连接器坐标写法 | JDK 建议 | 备注 |
|---|---|---|---|
| 1.14 及以下 | flink-connector-kafka_2.12:1.14.x | JDK 8 | 老接口，FlinkKafkaConsumer 还能用 |
| 1.15 / 1.16 | flink-connector-kafka:1.15.x/1.16.x | JDK 8/11 | 新版 KafkaSource 可用 |
| 1.17 / 1.18 | flink-connector-kafka:3.1.0-1.17等 | JDK 11 | 客户端兼容范围大，推荐新工程选这档 |
这里有个容易忽略的点：单独引kafka-clients时版本不能低于 Flink 连接器自带的那个，否则运行时会报UnsupportedVersionException，消息解析全乱。我一般会直接依赖连接器提供的客户端版本，不单独引 kafka-clients，除非确实需要新版 API。选型别追新，Flink 1.18 出了但生态插件跟进有滞后，连 Hive、JDBC 这类连接器都以 1.17 为基准更新得更勤。
2.2 用 docker-compose 起一个 Kafka 集群的最小配置
本地验证不需要真的搭三台机器，一个 Kafka 容器足够，但要注意：新版 Kafka 用了 KRaft 模式，不需要单独的 ZooKeeper 容器，很多老教程还是双容器方案，照抄会白占资源。下面这份 compose 我在多个环境用过，拉起来就能被 Flink 访问，还带一个 Kafka UI 便于看消息和 lag。
services: kafka: image: apache/kafka:3.7.0 container_name: kafka ports: - "9092:9092" environment: KAFKA_NODE_ID: 1 KAFKA_PROCESS_ROLES: broker,controller KAFKA_LISTENERS: PLAINTEXT://:9092,CONTROLLER://:9093 KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://localhost:9092 KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT KAFKA_CONTROLLER_QUORUM_VOTERS: 1@localhost:9093 KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1 KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 1 KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 1 KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS: 0 KAFKA_NUM_PARTITIONS: 3 kafka-ui: image: provectuslabs/kafka-ui:latest container_name: kafka-ui ports: - "8080:8080" environment: KAFKA_CLUSTERS_0_NAME: local KAFKA_CLUSTERS_0_BOOTSTRAPSERVERS: kafka:9092 depends_on: - kafka配置含义说一下：KAFKA_NUM_PARTITIONS设为 3，是为后面测试 Flink 并行度大于 1 时消息是否打散做准备；ADVERTISED_LISTENERS必须写localhost:9092，否则容器里启动成功了，宿主机上的 Flink 却连不上；GROUP_INITIAL_REBALANCE_DELAY_MS设 0 是让消费组尽快开始分配分区，测试时不用等那几秒的延迟。跑docker compose up -d后，先到 8080 端口看一眼 topic 是否存在，再用命令行生产几条数据，确认链路通再往下走。
2.3 pom.xml 依赖：四个必加的坐标与常见冲突
拿到 demo 工程第一件事是看 pom.xml，不是看代码。以下是我在 Flink 1.17 工程里最常用的最小依赖集，能覆盖从读取到执行的完整链路。
<properties> <flink.version>1.17.2</flink.version> <maven.compiler.source>11</maven.compiler.source> <maven.compiler.target>11</maven.compiler.target> </properties> <dependencies> <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> <scope>provided</scope> </dependency> <dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-connector-kafka</artifactId> <version>3.1.0-1.17</version> </dependency> <dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-table-common</artifactId> <version>${flink.version}</version> <scope>provided</scope> </dependency> </dependencies>flink-streaming-java提供 DataStream API，flink-clients负责本地执行环境，flink-connector-kafka是读取 Kafka 的核心，单独列出来是因为它的版本规则和 Flink 本体不同步。provided作用域要注意：在 IDEA 里直接运行 main 方法时，provided 的类默认不在运行 classpath 里，必须在 Run 配置里勾上 “Include dependencies with provided scope”，很多人第一次跑报ClassNotFoundException就是这个原因。
3. 读 Kafka 的三种写法：从 DataStream API 到 Flink SQL
3.1 最简实现：KafkaSource 与反序列化器
先从一个能直接跑通的最小例子开始。Flink 1.15 之后官方推荐用KafkaSource，老接口FlinkKafkaConsumer虽然还在，但已经标了 deprecated，新工程别再用。这个例子读取一个 topic 的字符串消息，打印到控制台。
import org.apache.flink.api.common.eventtime.WatermarkStrategy; import org.apache.flink.api.common.serialization.SimpleStringSchema; import org.apache.flink.connector.kafka.source.KafkaSource; import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer; import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment; public class KafkaReadDemo { public static void main(String[] args) throws Exception { StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); // 开启 checkpoint，Kafka offset 的提交依赖它 env.enableCheckpointing(5000); KafkaSource<String> source = KafkaSource.<String>builder() .setBootstrapServers("localhost:9092") .setTopics("demo-topic") .setGroupId("flink-demo-group") .setStartingOffsets(OffsetsInitializer.earliest()) .setValueOnlyDeserializer(new SimpleStringSchema()) .build(); env.fromSource(source, WatermarkStrategy.noWatermarks(), "kafka-source") .map(value -> "收到: " + value) .print(); env.execute("flink-read-kafka-demo"); } }这段代码有三个参数需要解释。setStartingOffsets(OffsetsInitializer.earliest())表示消费组第一次连上 topic 时从头开始读，如果你希望只消费新消息，改成latest()；setGroupId必须显式指定，Flink 靠它记录消费位点，不写会直接报错；setValueOnlyDeserializer(new SimpleStringSchema())只处理消息的 value 部分，如果你的消息里 key 也承载业务含义，需要换成setDeserializer并实现KafkaRecordDeserializationSchema，单纯用 value 是绝大多数 JSON 消息的默认场景。
3.2 带 checkpoint 的写法：状态与 Exactly-Once 的关系
没有 checkpoint 的 KafkaSource 是不可靠的，至少对这个标题下的生产场景来说不可用。Flink 的 Kafka 连接器不会在消费到一条消息时就立刻提交 offset，而是把提交动作绑定在 checkpoint 完成时——checkpoint 成功，offset 才提交。这样设计是为了避免“数据处理了但 offset 没提交，重启后重复消费”和“offset 提交了但数据处理失败，重启后丢数据”。
CheckpointConfig config = env.getCheckpointConfig(); config.setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE); config.setMinPauseBetweenCheckpoints(500); config.setCheckpointTimeout(60000); config.setMaxConcurrentCheckpoints(1); config.setExternalizedCheckpointCleanup( ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);这段配置里，EXACTLY_ONCE是 Flink 状态层面的保证，它不等于端到端精确一次——Kafka 到 Flink 这一段靠 checkpoint 能精确，但 Flink 写到 MySQL 或 Hive 那段还需要下游配合幂等。setMinPauseBetweenCheckpoints(500)是两次 checkpoint 之间的最小间隔，防止频繁快照把磁盘打满；RETAIN_ON_CANCELLATION是“后悔药”，作业取消后 checkpoint 文件保留，下次启动可以从最近一次快照恢复，调试血泪经验告诉我这个一定要开。
3.3 Flink SQL 方式：用 DDL 把 Kafka 当表读
如果你的场景是快速验证数据 or 不想写 Java 代码，Flink SQL 是更轻的路线。用 Table API 或 sql-client 执行下面这段 DDL，就把 Kafka topic 声明成了一张临时表，之后可以像查数据库一样查它。
CREATE TABLE kafka_demo ( `id` BIGINT, `name` STRING, `ts` TIMESTAMP(3), WATERMARK FOR `ts` AS `ts` - INTERVAL '5' SECOND ) WITH ( 'connector' = 'kafka', 'topic' = 'demo-topic', 'properties.bootstrap.servers' = 'localhost:9092', 'properties.group.id' = 'flink-sql-demo', 'scan.startup.mode' = 'earliest-offset', 'format' = 'json' );SELECT `id`, COUNT(*) AS cnt FROM kafka_demo GROUP BY `id`;SQL 写法的关键参数在 WITH 子句里：scan.startup.mode支持earliest-offset、latest-offset、group-offsets，语义和 DataStream API 里的setStartingOffsets一一对应；format必须是消息真正的序列化格式，JSON 和 CSV 最常用，如果消息里带 schema 则选debezium-json或canal-json。这里有个坑：SQL 方式默认把 Kafka 当无界流，GROUP BY会触发状态无限增长，生产环境必须配合ttl或窗口使用，demo 里无所谓，但要知道边界在哪。
4. 必调参数与消费稳定性：延迟、吞吐、消息顺序三者怎么平衡
4.1 从 kafka 消息延迟高说起：fetch、linger 与并行度的关系
很多人遇到 kafka 消息延迟高，第一反应是加 Kafka 节点或调大fetch.min.bytes，这往往跑偏。延迟高的根因更多在消费端的 Flink 作业并行度上：Kafka 一个分区只会被 Flink source 的一个并行子任务消费，如果你 topic 有 12 个分区，而 Flink 作业并行度只有 3，那理论上再快也只能同时拉 3 个分区的数据，消息全堵在 broker 上等分配。先查并行度，再看 broker 的fetch.max.wait.ms。
| 参数 | 默认值 | 作用 |
|---|---|---|
fetch.min.bytes | 1 | broker 攒够多少字节才返回，调大提升吞吐但抬高延迟 |
fetch.max.wait.ms | 500 | 等数据的最长时间，攒不够字节也会返回 |
max.poll.records | 500 | 单次 poll 返回的最大记录数 |
max.partition.fetch.bytes | 1048576 | 单个分区单次拉取上限 |
Flink 连接器的properties里可以直接传 Kafka 参数，例如.setProperty("fetch.min.bytes", "1024")。实践上，延迟敏感项目把fetch.min.bytes保持默认，靠并行度吃吞吐；吞吐敏感项目才调大它。有一类延迟是“假延迟”，消息量小、broker 一直等数据攒够字节，fetch.max.wait.ms到了上限才返回，这时调小fetch.max.wait.ms能见效。
4.2 消费端多线程与消息顺序性：最大化吞吐的代价
kafka 消费端多线程如何保证消息顺序性是个高频问题，答案很扫兴：Flink 里单并行子任务内部严格按 offset 顺序处理，一旦并行度大于 1，或者你在单并行里再开多线程做异步处理，顺序就打散了。顺序不是免费的，它和吞吐天然冲突。
我一般给业务方的建议就两条：一是需要全局有序的场景，并行度设 1，从源头放弃吞吐；二是按 key 局部有序的场景，用keyBy(key)让同一个 key 的消息落到同一并行实例，这是性价比最高的折中。比如订单变更流，只需要同一个订单号的消息有序，keyBy(orderId)就能在保住顺序的同时把不同订单分散到多个并行。注意keyBy之后你就不能依赖 Kafka 本身的 offset 顺序了，算子内部是按键分组后的顺序。
4.3 一套能直接抄的调参基线
基于上面这些约束，我给出一个通用性较强的初始配置，适合日均千万级消息、允许秒级延迟的项目。不是万能公式，但是一个能跑的起点。
| 配置项 | 推荐值 | 理由 |
|---|---|---|
| Flink 并行度 | topic 分区数的 1 到 2 倍 | 保证每个分区都有对应消费线程，留冗余应对故障 |
enable.auto.commit | false | 避免 Kafka 自动提交和 checkpoint 提交打架 |
| checkpoint 间隔 | 30 到 60 秒 | 太频繁磁盘压力大，太慢重启恢复丢得多 |
max.poll.records | 500 到 1000 | 超过处理能力会触发 rebalance，反而降吞吐 |
| restart-strategy | fixed-delay，3 次，间隔 10 秒 | 防止脏数据让作业无限重启 |
fetch.max.wait.ms | 500 | 延迟和吞吐的平衡点 |
这套基线跑稳定后再逐步压测调优，一次只动一个参数。我见过最典型的翻车就是把所有参数一次性拉满，结果是吞吐没上去，checkpoint 超时、rebalance 频繁，最后把锅甩给 Kafka——这锅 Kafka 不背。
5. 避坑：打包、序列化、连接器错误与数据积压的排查清单
5.1 现象：ClassNotFound / NoClassDefFound，作业提交后直接失败
原因：本地 IDEA 跑通了，打包传到集群却起不来，十有八九是连接器依赖没有打进 jar 包。providedscope 的依赖在集群里有，但flink-connector-kafka和kafka-clients这些 runtime 依赖必须跟着作业包走，默认 maven package 不会打入。解决：用maven-shade-plugin打成 fat jar，并把连接器依赖正常打入。
<plugin> <groupId>org.apache.maven.plugins</groupId> <artifactId>maven-shade-plugin</artifactId> <version>3.5.1</version> <executions> <execution> <phase>package</phase> <goals><goal>shade</goal></goals> <configuration> <createDependencyReducedPom>false</createDependencyReducedPom> </configuration> </execution> </executions> </plugin>检查打包是否成功有个土办法：看最终 jar 的大小。只有几十 KB 的肯定没带上连接器，正常带 Kafka 连接器的 fat jar 至少几 MB 起步。
5.2 现象：Flink 的 JDBC 连接器异常——依赖与运行时类加载器冲突
原因：作业里同时用了flink-connector-jdbc和自己单独引的 MySQL 驱动，版本对不上，或两个驱动 jar 都进了 classpath，类加载器优先加载了旧版本。这类报错花样很多，最常见的是ClassNotFoundException: com.mysql.cj.jdbc.Driver或No suitable driver found。解决：只保留一个驱动版本，且用 shade 插件把驱动 relocate 到一个私有命名空间，避免和集群自带的类冲突。
<relocation> <pattern>com.mysql.cj</pattern> <shadedPattern>shaded.com.mysql.cj</shadedPattern> </relocation>另一种更干净的思路：把 JDBC 连接器和驱动直接放到 Flink 的lib目录，作业包里不再打入，这样类加载器冲突面最小。按 Flink 官方默认的 child-first 类加载策略，放 lib 里的类优先级高于作业包，能避免大部分冲突。
5.3 现象：Sink Hive 表数据不入表，作业不报错但查不到数据
原因：表面是“没写进去”，实际常见原因有三个——Hive 方言没开启、分区字段类型或格式不匹配、写入模式不对。我遇到过最隐蔽的是分区字段定义dt STRING，而 Flink SQL 里写的是TIMESTAMP，两者在 Catalog 里映射成不同类型，Hive 建了分区目录但查SELECT *看不到。解决：开 Flink SQL 时先执行set table.sql-dialect=hive，写在INSERT INTO t PARTITION(dt='20240601')时显式指定分区值，避免让 Flink 推断。还要确认 HiveCatalog 里表 schema 的类型和 Flink 端完全一致，别依赖隐式转换。
5.4 现象：反序列化失败导致作业无限重启，offset 卡住不前进
原因：topic 里混入脏数据，比如本该是 JSON 的消息夹了一行纯文本，SimpleStringSchema还好说，一旦自定义解析抛异常，Flink 默认的失败策略是不断重启，每次重启从前一个成功 checkpoint 恢复，于是那条脏数据被反复消费、反复失败，形成一个“永远过不去的坎”。解决：配置重启策略限制次数，更重要是加侧输出流，把解析失败的数据单独接走，不做主链路。
SingleOutputStreamOperator<String> parsed = env.fromSource(...) .process(new ProcessFunction<String, String>() { @Override public void processElement(String value, Context ctx, Collector<String> out) { try { JSONObject obj = JSON.parseObject(value); out.collect(obj.getString("id")); } catch (Exception e) { // 脏数据走侧输出，不打断主流程 ctx.output(new OutputTag<String>("dirty") {}, value); } } });代码里OutputTag是侧输出流的标签，脏数据在这里被分流，restart-strategy即使触发了，恢复后也能跳过这条消息继续向后消费。这个模式比“无限重启直到人工删 offset”体面得多，也一直是生产上我优先推荐的容错姿势。
5.5 现象：Kafka 消息延迟高——消费端反压与空闲分区
原因：这一类要区分“源头慢”和“下游慢”。常见的是 sink 写入数据库太慢，反压一路传回 source，source 不敢继续拉消息，lag 自然上涨；另一类是 source 并行度大于分区数，有一部分子任务分配不到分区，lag 分布极度不均。解决：先看监控，确认是 source 侧还是 sink 侧背压，再对症下药——sink 侧慢就优化写入（批量、异步、连接池），source 侧不均就把并行度调到和分区数一致或更低。用 Kafka UI 或kafka-consumer-groups.sh查 lag 分布，一眼能分辨这两种情况：所有分区 lag 一起涨是下游反压，只有部分分区涨是分配不均。
6. 把一条生产级链路串起来：读取、转换、写入带状态的目标端
前面讲的都是单点能力，最后把它们串成一条完整链路：Kafka 消费 → 按用户分组统计会话次数 → 写入 MySQL。这个例子几乎覆盖了 flink 读取 kafka 数据 的全部核心动作，也是你能从这个 zip 方向里带走的最有价值的东西。
public class KafkaToMysqlDemo { public static void main(String[] args) throws Exception { StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); env.enableCheckpointing(30000); KafkaSource<String> source = KafkaSource.<String>builder() .setBootstrapServers("localhost:9092") .setTopics("user-click-topic") .setGroupId("demo-group") .setStartingOffsets(OffsetsInitializer.latest()) .setValueOnlyDeserializer(new SimpleStringSchema()) .build(); env.fromSource(source, WatermarkStrategy.noWatermarks(), "kafka-source") .map(value -> JSON.parseObject(value)) .keyBy(json -> json.getString("userId")) .map(new CountAgg()) .addSink(new MysqlSink()); env.execute("kafka-to-mysql-demo"); } }keyBy(userId)保证同一个用户的所有点击落到同一并行实例，CountAgg里可以用ValueState保存累计值，但注意状态会在 checkpoint 里持久化，别只把它当普通变量。MysqlSink 里我强烈建议用主键 upsert 而不是 insert，这样 Kafka 重复消费时，写入 MySQL 是幂等的，端到端精确一次才有落点。这个工程方向值得做，而且越早把“状态 + checkpoint + 幂等写入”这套组合想明白，后面接 Hive、接 Iceberg、接 CDC 都只是换连接器，骨架不会变。
我现在接手任何人的老作业，第一件事永远是看 checkpoint 配置和 sink 有没有幂等，这两点过关，作业基本能睡安稳觉。希望帮到你。
本文还有配套的精品资源，点击获取
