一、简介

flink 自定义实时数据源使用流处理比较简单，比如 Kafka、MQ 等，如果使用 MySQL、redis 批处理也比较简单

如果需要定时加载数据作为 flink 数据源使用流处理，比如定时从 mysql 或者 redis 获取一批数据，传入 flink 做处理，如下简单实现

二、pom.xml 文件

注意 flink 好多包从 1.15.0 开始不需要指定 Scala 版本，内部自带

下面 pom 文件有 flink 两个版本 1.16.0 和 1.12.7（Scala:2.12）

4.0.0com.yeflink-study0.1jarFlink Quickstart JobUTF-81.16.01.82.12${target.java.version}${target.java.version}2.17.1apache.snapshotsApache Development Snapshot Repositoryhttps://repository.apache.org/content/repositories/snapshots/falsetrueorg.apache.flinkflink-streaming-java${flink.version}providedorg.apache.flinkflink-clients${flink.version}provided
org.apache.flinkflink-connector-files${flink.version}org.apache.logging.log4jlog4j-slf4j-impl${log4j.version}runtimeorg.apache.logging.log4jlog4j-api${log4j.version}runtimeorg.apache.logging.log4jlog4j-core${log4j.version}runtimeorg.apache.maven.pluginsmaven-compiler-plugin3.1${target.java.version}${target.java.version}org.apache.maven.pluginsmaven-shade-plugin3.1.1packageshadefalseorg.apache.flink:flink-shaded-force-shadingcom.google.code.findbugs:jsr305org.slf4j:*org.apache.logging.log4j:**:*META-INF/*.SFMETA-INF/*.DSAMETA-INF/*.RSAcom.ye.DataStreamJoborg.eclipse.m2elifecycle-mapping1.0.0org.apache.maven.pluginsmaven-shade-plugin[3.1.1,)shadeorg.apache.maven.pluginsmaven-compiler-plugin[3.1,)testCompilecompile

三、自定义数据源

使用 Timer 定时任务（当然也可以使用线程池 Executors）自定义数据源，每过五秒随机生成一串字符串

public class TimerSinkRich extends RichSourceFunction {private  ConcurrentLinkedQueue queue = new ConcurrentLinkedQueue<>();private boolean flag = true;private Timer timer;private TimerTask timerTask;@Overridepublic void open(Configuration parameters) throws Exception {super.open(parameters);timerTask = new TimerTask() {@Overridepublic void run() {// 可以在这块获取 MySQL、redis 等连接并查询数据Random random = new Random();StringBuilder str = new StringBuilder();for (int i = 0; i < 10; i++) {char ranLowLetter = (char) ((random.nextInt(26) + 97));str.append(ranLowLetter);}queue.add(str.toString());}};timer = new Timer();// 延时和执行周期参数可以通过构造方法传递timer.schedule(timerTask,1000,5000);}@Overridepublic void run(SourceContext ctx) throws Exception {while (flag){if(queue.size()>0){ctx.collect(queue.remove());}}}@Overridepublic void cancel() {if(null!=timer) timer.cancel();if(null!=timerTask) timerTask.cancel();// 撤销任务时，flink 默认 30 s（不同 flink 版本可能不同）尝试关闭数据源，关闭失败 TaskManager 不能释放 slot，最终导致失败if(queue.size()<=0) flag = false;}
}

四、flink 加载数据源并启动

public class TimerSinkStreamJob {public static void main(String[] args) throws Exception {StreamExecutionEnvironment executionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment();executionEnvironment.setParallelism(1);DataStreamSource streamSource = executionEnvironment.addSource(new TimerSinkRich());streamSource.print();executionEnvironment.execute("TimerSinkStreamJob 定时任务打印数据");}
}

本地测试成功

五、上传 flink 集群

1、flink 1.16.0

启动成功

撤销任务成功

solt 也成功释放

2、flink 1.12.7

启动成功

撤销任务当然也没问题，同样能正常释放 slot

当然你也可以不要 open() 方法

public class DiySinkRich extends RichSourceFunction {private TimerTask timerTask;private Timer timer;private boolean flag = true;private ConcurrentLinkedQueue queue = new ConcurrentLinkedQueue<>();@Overridepublic void run(SourceFunction.SourceContext ctx) throws Exception {timerTask = new TimerTask() {@Overridepublic void run() {Random random = new Random();StringBuilder str = new StringBuilder();for (int i = 0; i < 10; i++) {char ranLowLetter = (char) ((random.nextInt(26) + 97));str.append(ranLowLetter);}queue.add(str.toString());}};timer = new Timer();timer.schedule(timerTask, 1000, 5000);while (flag) {if (queue.size() > 0) {ctx.collect(queue.remove());}}}@Overridepublic void cancel() {if (timer != null) timer.cancel();if (timerTask != null) timerTask.cancel();if (queue.size() == 0) flag = false;}
}

以上就是 flink 定时加载数据源的简单实例