说明

有时，flink内置的数据源不能满足需求，我们还可以自定义数据源，有如下方式：

• 实现 SourceFunction 接口
• 实现 ParallelSourceFunction 接口
• 继承 RichParallelSourceFunction

SourceFunction

SourceFunction 是 非并行的，所以不能指定并行度，执行下面代码会报错：

env.addSource(new SourceFunction() ).setParallelism(2)

源码

public interface SourceFunction<T> extends Function, Serializable {

    void run(SourceContext<T> ctx) throws Exception;
    void cancel();
}

方法解释：

• run()：调用 run() 方法产生数据
• cancel()：设置变量，用于终止 run() 方法执行

泛型 SourceFunction<T> ：表示数据源中的数据类型

应用场景

• 网络socket是串行的，适合该接口，所以 SocketTextStreamFunction 就是实现的 SourceFunction

• 根据 集合创建数据源，底层也是实现的 SourceFunction

ParallelSourceFunction

ParallelSourceFunction 是 并行 的，所以能指定并行度

源码

public interface ParallelSourceFunction<OUT> extends SourceFunction<OUT> {
}

该接口没有方法

泛型 ParallelSourceFunction<OUT> ：表示数据源中的数据类型

RichParallelSourceFunction

RichParallelSourceFunction 不仅实现了ParallelSourceFunction，还继承了 AbstractRichFunction

源码

public abstract class RichParallelSourceFunction<OUT> extends AbstractRichFunction
        implements ParallelSourceFunction<OUT> {

    private static final long serialVersionUID = 1L;
}

方法解释：

由于继承了 AbstractRichFunction 类，所以有以下方法：

• open()：初始化时执行一次，如：连接mysql
• close()：在source关闭的时执行一次，如：关闭mysql连接

泛型 RichParallelSourceFunction<OUT> ：表示数据源中的数据类型

应用场景

内置的 kafka连接器 FlinkKafkaConsumer，底层代码就是继承了 RichParallelSourceFunction 类

参考

https://www.cnblogs.com/xiexiandong/p/12770187.html

原文出处：/show_1IX42gf4g3Mt.html