大数据 -114 Flink DataStreamAPI 从 SourceFunction 到 RichSourceFunction 源函数的增强与实战

2025-10-03
山东
本文字数：4627 字
阅读完需：约 15 分钟

点一下关注吧！！！非常感谢！！持续更新！！！

🚀 AI 篇持续更新中！（长期更新）

AI 炼丹日志-31- 千呼万唤始出来 GPT-5 发布！“快的模型 + 深度思考模型 + 实时路由”，持续打造实用 AI 工具指南！📐🤖

💻 Java 篇正式开启！（300 篇）

目前 2025 年 09 月 29 日更新到：Java-136 深入浅出 MySQL Spring Boot @Transactional 使用指南：事务传播、隔离级别与异常回滚策略 MyBatis 已完结，Spring 已完结，Nginx 已完结，Tomcat 已完结，分布式服务正在更新！深入浅出助你打牢基础！

📊 大数据板块已完成多项干货更新（300 篇）：

包括 Hadoop、Hive、Kafka、Flink、ClickHouse、Elasticsearch 等二十余项核心组件，覆盖离线+实时数仓全栈！大数据-278 Spark MLib - 基础介绍机器学习算法梯度提升树 GBDT 案例详解

章节内容

上节完成了如下的内容：

FlinkDataStreamAPI 自定义输入源
非并行源介绍与代码
并行源介绍与代码

Rich 并行源

基本介绍

在 Apache Flink 的流处理框架中，RichSourceFunction 是一种功能增强型的源函数（Source Function），它为开发者提供了比普通 SourceFunction 更强大的功能集。RichSourceFunction 主要应用于数据源的定义和实现，能够支持更复杂的业务场景和数据处理需求。

核心特性

生命周期方法：RichSourceFunction 提供了完整的生命周期管理能力，包括：
open()：在任务启动时执行初始化操作
close()：在任务结束时执行清理工作
run()：定义数据生成逻辑的主体方法
cancel()：用于优雅地终止数据生成
状态管理：通过继承 RichFunction 接口，可以访问：
键控状态（Keyed State）
算子状态（Operator State）
状态后端（State Backend）配置
运行时上下文：可以通过 getRuntimeContext() 方法获取：
任务执行环境信息
并行度设置
任务实例编号
分布式缓存访问
配置访问：支持从 Flink 的配置系统中读取参数，实现更灵活的配置方式。

继承关系

RichSourceFunction 的类继承关系如下：

SourceFunction    ↑AbstractRichFunction    ↑RichSourceFunction

复制代码

典型应用场景

需要状态管理的源：例如从断点续传的消息队列中消费数据
需要访问分布式缓存的源：比如在数据生成时查询外部参考数据
需要复杂初始化的源：如建立数据库连接、加载配置文件等
需要精确控制生命周期的源：如资源密集型的数据源

示例代码

public class CustomRichSource extends RichSourceFunction<String> {    private transient Connection dbConnection;        @Override    public void open(Configuration parameters) {        // 初始化数据库连接        dbConnection = DriverManager.getConnection("jdbc:mysql://...");    }        @Override    public void run(SourceContext<String> ctx) {        // 从数据库读取数据并发出        while (isRunning) {            ResultSet rs = dbConnection.executeQuery(...);            while (rs.next()) {                ctx.collect(rs.getString(1));            }        }    }        @Override    public void cancel() {        // 设置取消标志        isRunning = false;    }        @Override    public void close() {        // 关闭数据库连接        if (dbConnection != null) {            dbConnection.close();        }    }}

复制代码

与普通 SourceFunction 的区别

在实际开发中，RichSourceFunction 更适合处理需要复杂逻辑或资源管理的源操作，而简单的数据源可以直接使用基本的 SourceFunction 实现。

主要特点

生命周期方法：RichSourceFunction 提供了 open() 和 close() 方法，分别在作业开始时和结束时调用。这允许你在数据读取前进行初始化操作（如打开连接、加载配置），以及在作业结束时进行清理工作（如关闭连接、释放资源）。
访问运行时上下文：通过 getRuntimeContext() 方法，RichSourceFunction 可以访问 Flink 的运行时上下文，获取并行度信息、任务名称、指标管理器，以及与状态相关的操作。
状态管理：RichSourceFunction 可以结合 Flink 的状态管理机制，保存和恢复状态。这对于需要在流处理中维护中间状态的源函数非常有用，尤其是在故障恢复时，状态可以帮助恢复到故障前的状态。
并行执行：与普通的 SourceFunction 类似，RichSourceFunction 也可以通过设置并行度来并行执行，这使得它可以处理大规模的数据源。

状态管理

RichFunction 与 Flink 的状态管理系统高度集成，提供了一套完整的机制来维护和管理操作符的中间状态。这种状态管理能力是 Flink 实现精确一次（exactly-once）语义的核心保障。Flink 的状态后端（State Backend）负责状态的存储和访问，支持多种存储方式包括内存、文件系统和 RocksDB。

Flink 支持四种主要类型的状态，每种状态都有其特定的应用场景：

ValueState

适用于需要保存单个值的场景：

典型应用：计数器（如统计事件数量）、标志位（如处理状态标记）
示例：在数据流处理中统计总交易金额

ValueState<Double> totalAmount = getRuntimeContext()    .getStateProsition(new ValueStateDescriptor<>("total-amount", Types.DOUBLE));

复制代码

ListState

适用于需要保存多个值的场景：

典型应用：窗口计算中的中间结果、历史记录存储
示例：存储最近 N 次登录失败的 IP 地址

ListState<String> failedIPs = getRuntimeContext()    .getListState(new ListStateDescriptor<>("failed-ips", Types.STRING));

复制代码

MapState

适用于需要维护键值对的场景：

典型应用：复杂数据关联、用户画像特征存储
示例：维护用户 ID 到其行为特征的映射

MapState<String, UserBehavior> userBehaviors = getRuntimeContext()    .getMapState(new MapStateDescriptor<>("user-behaviors", Types.STRING, Types.POJO(UserBehavior.class)));

复制代码

ReducingState

适用于需要持续聚合数据的场景：

典型应用：实时计算平均值、最大值等聚合指标
示例：持续计算移动平均温度

ReducingState<Double> avgTemperature = getRuntimeContext()    .getReducingState(new ReducingStateDescriptor<>("avg-temp", new AverageReducer(), Types.DOUBLE));

复制代码

状态的生命周期管理包括状态的注册、初始化、更新和清除。Flink 还提供了状态 TTL（Time-To-Live）配置，可以自动清理过期的状态数据。在实际应用中，需要根据业务需求选择合适的状态类型，并考虑状态大小和访问频率对性能的影响。

示例代码

以下是一个使用 RichParallelSourceFunction 的简单示例，展示了如何在 Flink 中实现一个并行的、具有生命周期管理的源函数：

import org.apache.flink.configuration.Configuration;import org.apache.flink.streaming.api.functions.source.RichParallelSourceFunction;
public class MyRichParallelSource extends RichParallelSourceFunction<String> {
    private volatile boolean isRunning = true;
    @Override    public void open(Configuration parameters) throws Exception {        // 在任务开始时执行初始化操作        System.out.println("Task " + getRuntimeContext().getTaskName() + " is starting.");    }
    @Override    public void run(SourceContext<String> ctx) throws Exception {        // 模拟数据流的产生        while (isRunning) {            synchronized (ctx.getCheckpointLock()) {                ctx.collect("Data from task " + getRuntimeContext().getIndexOfThisSubtask());            }            Thread.sleep(1000);        }    }
    @Override    public void cancel() {        isRunning = false;    }
    @Override    public void close() throws Exception {        // 在任务结束时执行清理操作        System.out.println("Task " + getRuntimeContext().getTaskName() + " is closing.");    }}

复制代码

代码解析

open() 方法：在任务开始时调用，适用于进行连接初始化、参数设置等操作。在这个方法中，你可以访问 Flink 的配置和运行时上下文。
run() 方法：实现数据源的核心逻辑，这个方法会在源函数启动后被调用。可以使用 ctx.collect() 方法将生成的数据发送到下游处理。
cancel() 方法：用于取消任务。当作业被取消或停止时，Flink 会调用这个方法，可以在这里做一些清理工作或者安全地停止数据生成。
close() 方法：在任务结束时调用，用于释放资源和进行清理操作。

注意事项

状态一致性：在并行源中，如果需要维护状态，一定要注意状态的一致性和恢复机制，确保在作业恢复时可以正确地恢复数据源的状态。
并行度设置：RichParallelSourceFunction 作为并行源，可以通过 setParallelism 方法设置并行度，确保根据任务的需求合理分配并行实例的数量。

RichParallelSource

package icu.wzk;
import org.apache.flink.streaming.api.functions.source.RichParallelSourceFunction;
public class RichParallelSourceRich extends RichParallelSourceFunction<String> {
    private long count = 1L;    private boolean running = true;
    @Override    public void run(SourceContext<String> ctx) throws Exception {        while (running) {            count ++;            ctx.collect(String.valueOf(count));            Thread.sleep(1000);        }    }
    @Override    public void cancel() {        running = false;    }}

复制代码

RichParallelSourceTest

package icu.wzk;
import org.apache.flink.streaming.api.datastream.DataStreamSource;import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment;
public class RichParallelSourceRichTest {
    public static void main(String[] args) {        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();        DataStreamSource<String> data = env.getJavaEnv().addSource(new RichParallelSourceRich());        data.print();        env.execute("RichParallelSourceRichTest");    }
}

复制代码

运行结果

3> 105> 108> 106> 102> 104> 107> 101> 106> 115> 118> 112> 113> 114> 117> 111> 112> 123> 12...

复制代码

控制台输出结果如下所示：

为什么 Rich 类使用广泛

生命周期管理：Rich 类提供了 open() 和 close() 方法，允许开发者在任务开始和结束时执行初始化和清理操作。这对于需要设置资源（如数据库连接、文件读写、外部服务连接）的操作非常有用。
运行时上下文访问：通过 getRuntimeContext()，Rich 类可以访问任务的并行度信息、任务名称、子任务索引、状态管理等。对于需要根据任务上下文调整行为或需要跨并行实例共享状态的场景，这些信息是至关重要的。
状态管理：RichFunction 可以方便地与 Flink 的状态管理结合使用。在状态丰富的应用场景（如需要维护中间计算结果、计数器、缓存等）的流处理中，Rich 类显得非常有用。
性能监控：Rich 类允许开发者在 open() 方法中注册 Flink 的度量指标（Metrics），帮助监控和优化作业的性能。

什么时候不用 Rich 类

简单操作：如果你只是需要进行简单的转换或过滤操作，没有复杂的初始化、状态管理或清理需求，那么 Rich 类的额外功能可能并不必要。
高性能要求的场景：在一些对性能要求极高的场景中，尽量减少复杂的操作和额外的上下文访问，直接使用轻量级的 MapFunction、FilterFunction 等可能会有更好的性能表现。

发布于: 刚刚阅读数: 2

原文链接:【http://xie.infoq.cn/article/8c8a80ac3994107c3c9c68fea】。文章转载请联系作者。

武子康

关注

永远好奇无限进步 2019-04-14 加入

Hi, I'm Zikang，好奇心驱动的探索者 | INTJ / INFJ 我热爱探索一切值得深究的事物。对技术、成长、效率、认知、人生有着持续的好奇心和行动力。坚信「飞轮效应」，相信每一次微小的积累，终将带来深远的改变。

发布

暂无评论

创作场景

大数据 -114 Flink DataStreamAPI 从 SourceFunction 到 RichSourceFunction 源函数的增强与实战

点一下关注吧！！！非常感谢！！持续更新！！！

🚀 AI 篇持续更新中！（长期更新）

💻 Java 篇正式开启！（300 篇）

📊 大数据板块已完成多项干货更新（300 篇）：

章节内容

Rich 并行源

基本介绍

核心特性

继承关系

典型应用场景

示例代码

与普通 SourceFunction 的区别

主要特点

状态管理

ValueState

ListState

MapState

ReducingState

示例代码

代码解析

注意事项

RichParallelSource

RichParallelSourceTest

运行结果

为什么 Rich 类使用广泛

什么时候不用 Rich 类

武子康

评论