flink定时器_flink定时器每隔

flink中可以实现每n秒执行一个方法的定时任务吗？使用Ja自己的定时作是失效的

是的，Apache Flink可以实现每n秒执行一个方法的定时任务。Flink提供了两种实现方式：

flink定时器_flink定时器每隔

flink定时器_flink定时器每隔

Timer Serv是Flink中的一种内置机制，可以基于时间戳触发回调函数。在使用Timer Serv时，可以通过ProcessingTimeServ指定定时器的触发时间，例如每n秒触发一次。以下是使用Timer Serv实现每5秒执行一次方法的示例代码：

public class TimerExample {

public static void main(String[] args) throws Exception {

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

env.addSource(new SourceFunction() {

public void run(SourceContext ctx) throws Exception {

while (true) {

ctx.collect("Hello, World!");

Thread.sleep(5000);

public void cancel() {

}}).addSink(new SinkFunction() {

public void invoke(String value) throws Exception {

System.out.println(value);

}});

env.execute("Timer Example");

在上面的示例代码中，源函数中每5秒发出一个，Sink函数中接收并打印出。使用Timer Serv时，需要注意：

Timer Serv是基于时间戳的，因此需要确保中包含时间戳信息。

Timer Serv是由Flink系统驱动的，因此需要在Flink作业中执行。

2.使用Ja的ScheduledExecutorServ

除了Timer Serv之外，还可以使用Ja的ScheduledExecutorServ来实现定时任务。以下是使用ScheduledExecutorServ实现每5秒执行一次方法的示例代码：

public class ScheduledExecutorExample {

public static void main(String[] args) throws Exception {

ScheduledExecutorServ executorServ = Executors.newScheduledThreadPool(1);public void run() {

executorServ.scheduleAtFixedRate(new Runnable() {

System.out.println("Hello, World!");

}}, 0, 5, TimeUnit.SECONDS);

Thread.sleep(60000);

executorServ.shutdown();

在上面的示例代码中，使用scheduleAtFixedRate方法指定每5秒执行一次方法。需要注意的是，使用ScheduledExecutorServ时，需要在Flink作业中执行。

import org.apache.flink.streaming.api.functions.source.SourceFunction;public class TimerTask {

public static void main(String[] 可以看到根据ProcessingTime和EventTime的不同，分别加入不同的queue队列中，其中如果是ProcessingTime的话，他还会判断当前queue中个元素的触发时间是否比当前加入的注册时间晚，如果晚于当前新加入的时间，则把下次触发时间改成当前的新加入的注册时间。args) throws Exception { StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

// 每秒执行一次 env.setStreamTimeCharacteristic(TimeCharacteristic.ProcessingTime); env.setParalleli(1);

env.addSource(new SourceFunction() {

public void run(SourceContext ctx) throws Exception { while (true) {

ctx.collect("Hello World");

Thread.sleep(1000);

} }

public void cancel() {

} }).print(); env.execute("TimerTask"); }}

flink中ProcessFunction的注册定时器功能

// 每秒钟发出一个消息

在flink的ProcessFunction中，我们可以注册定时器设定延迟多长时间后执行某类作，例如像这种：

context.timerServ().registerEventTimeTimer(context.timestamp() + 10000);

很好奇这种定时器内部是如何进行工作的，带着这种疑问我们来看看源码。

首先，在源码中，所有的定时器管理都是通过InternalTimerServImpl.ja这个类来实现的。

我们接下来看queue队列是如何实现的，processingTimeTimersQueue和ntTimeTimersQueue实现原理都是一样的，它的实现类是HeapPrior时间： 由数据源产生ityQueueSet.ja:

这里的逻辑简单概况就是先简单判断该注册时间是否有重复，如果没有重复就继续往里添加，再来看super.add(element)这个方法的实现：

重点看siftUp这个方法，这个方法实现的的就是堆排序并且还是小顶堆排序，先把新的定时器放到数组末尾，然后就进行小顶堆排序，永远把最小的元素（定时器）排到最前面，这样就最早触发。至此逻辑就很清楚了：添加定时器的时候，首先会判断是否有重复，然后进行小顶堆排序，把最小的定时器放到个。

接下来我们继续看定时器是如何触发的，先看InternalTimerServImpl.ja：

在收到watermark之后开始执行aanceWatermark方法，这时候从ntTimeTimersQueue中获得个定时器（之前加入的时候已经保证了弹出的个永远是时间最早的定时器）与当前watermark时间比较，如果小于watermark则取出该定时器执行onEventTime也就是ProcessFunction中的onTimer方法。

取出个定时器之后，会触发ntTimeTimersQueue中的小顶堆再次排序：

这里简单概况来说就是弹出个定时器，同时触发小顶堆再次排序，把数组中剩余的时间最小的定时器再次放到个位置.

processTime是依靠自身线程注册的定时器来触发的，processingTimeTimersQueue的逻辑与ntTimeTimersQueue一样，这里就不多讲了，当弹出个定时器执行的时候，会立即注册下一个定时器，保证下一个定时器顺利按时执行

总计一下，在ProcessFunction中，ntTime依靠watermark来触发，processTime依靠自身线程注册的定时器触发，两者都是在添加定时器的时候，把定时器放入队列里面进行小顶堆排序，把时间最小的定时器放到个位置，最早触发。

flink中ProcessFunction的注册定时器功能

在flink的ProcessFunction中，我们可以注册定时器设定延迟多长时间后执行某类作，例如像这种：

context.timerServ().registerEventTimeTimer(context.timestamp() + 10000);

很好奇这种定时器内部是如何进行工作的，带着这种疑问我们来看看源码。

首先，在源码中，所有的定时器管理都是通过InternalTimerSeSparkStreaming 是将流处理分成微批处理的作业， 的处理引擎是spark jobrvImpl.ja这个类来实现的。

我们接下来看queue队列是如何实现的，processingTimeTimersQueue和ntTimeTimersQueue实现原理都是一样的，它的实现类是HeapPriorityQueueSet.ja:

这里的逻辑简单概况就是先简单判断该注册时间是否有重复，如果没有重复就继续往里添加，再来看super.add(element)这个方法的实现：

重点看siftUp这个方法，这个方法实现的的就是堆排序并且还是小顶堆排序，先把新的定时器放到数组末尾，然后就进行小顶堆排序，永远把最小的元素（定时器）排到最前面，这样就最早触发。至此逻辑就很清楚了：添加定时器的时候，首先会判断是否有重复，然后进行小顶堆排序，把最小的定时器放到个。

接下来我们继续看定时器是如何触发的，先看InternalTimerServImpl.ja：

在收到watermark之后开始执行aanceWatermark方法，这时候从ntTimeTimersQueue中获得个定时器（之前加入的时候已经保证了弹出的个永远是时间最早的定时器）与当前watermark时间比较，如果小于watermark则取出该定时器执行onEventTime也就是ProcessFunction中的onTimer方法。

取出个定时器之后，import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;会触发ntTimeTimersQueue中的小顶堆再次排序：

这里简单概况来说就是弹出个定时器，同时触发小顶堆再次排序，把数组中剩余的时间最小的定时器再次放到个位置.

processTime是依靠自身线程注册的定时器来触发的，processingTimeTimersQueue的逻辑与ntTimeTimersQueue一样，这里就不多讲了，当弹出个定时器执行的时候，会立即注册下一个定时器，保证下一个定时器顺利按时执行

总计一下，在ProcessFunction中，ntTime依靠watermark来触发，processTime依靠自身线程注册的定时器触发，两者都是在添加定时器的时候，把定时器放入队列里面进行小顶堆排序，把时间最小的定时器放到个位置，最早触发。

Flink 原理详解

}}

Flink 是一个流处理框架，支持流处理和批处理，特点是流处理有限，可容错，可扩展，高吞吐，低延迟。

流处理是处理一条，立马下一个会从缓存中ntTime触发的定时器逻辑这里就讲完了，我们再看看processTime的触发逻辑，在InternalTimerServImpl.ja中：取出，在下一个进行计算

批处理是只有处理一批完成后，才会经过网络传输到下一个

流处理的优点是低延迟 批处理的优点是高吞吐

flink同时支持两种，flink的网络传输是设计固定的缓存块为单位，用户可以设置缓存块的超时值来决定换存块什么时候进行传输。 数据大于0 进行处理就是流式处理。

如果设置为无限大就是批处理模型。

Flink 集群包括 JobMar 和 TaskMar .

JobMar 主要负责调度 Job 并协调 Task 做 checkpoint，职责上很像 Storm 的 Nimbus。从 处接收到 Job 和 JAR 包 等资源后，会生成优化后的执行，并以 Task 的单元调度到各个 TaskMar 去执行。

TaskMar 在启动的时候就设置好了槽位数（Slot），每个 slot 能启动一个 Task，Task 为线程。从 JobMar 处接收需要 部署的 Task，部署启动后，与自己的上游建立 Netty 连接，接收数据并处理。

flink on yarn 是由client 提交 app到 RM 上， 然后RM 分配一个 AppMaster负责运行 Flink JobMar 和 Yarn AppMaster, 然后 AppMaster 分配 容器去运行 Flink TaskManger

Spark Streaming把实时输入数据流以时间片Δt （如1秒）为单位切分成块，Spark Streaming会把每块数据作为一个RDD，并使用RDD作处理每一小块数据。每个块都会生成一个Spark Job处理，然后分批次提交job到集群中去运行，运行每个 job的过程和真正的spark 任务没有任何区别。

JobScheduler, 负责 Job的调度通过定时器每隔一段时间根据Dstream的依赖关系生一个一个DAG图

ReceiverTracker负责数据的接收，管理和分配

ReceiverTracker在启动Receiver的时候他有ReceiverSupervisor,其实现是ReceiverSupervisorImpl, ReceiverSupervisor本身启 动的时候会启动Receiver，Receiver不断的接收数据，通过BlockGenerator将数据转换成Block。定时器会不断的把Block数据通会不断的把Block数据通过BlockMar或者WAL进行存储，数据存储之后ReceiverSupervisorlmpl会把存储后的数据的元数据Metadate汇报给ReceiverTracker，其实是汇报给ReceiverTracker中的RPC实体ReceiverTrackerEndpoin

spark on yarn 的cluster模式， Spark client 向RM提交job请求, RM会分配一个 AppMaster, driver 和 运行在AppMAster里， AM然后把Receiver作为一个Task提交给Spark Executor ， Receive启动接受数据，生成数据块，并通知Spark App, AM会根据数据块生成相应的Job, 并把Job 提交给空闲的 Executor 去执行。

1：需要关注流数据是否需要进行状态管理

3：对于小型的项目，并且需要低延迟的场景，建议使用storm

4：如果你的项目已经使用了spark，并且秒级别的实时处理可以满足需求的话，建议使用sparkStreaming

5：要求消息投递语义为 Exactly Once 的场景；数据量较大，要求高吞吐低延迟的场景；需要进行状态管理或窗口统计的场景，建议使用flink

Flink 提供的Api右 DataStream 和 DataSet ，他们都是不可变的数据，不可以增加删除中的元素， 通过 Source 创建 DataStream 和 DataSet

在创建运行时有：

Flink的每一个Operator称为一个任务， Operator 的每一个实例称为子任务，每一个任务在JVM线程中执行。可以将多个子任务链接成一个任务，减少上下文切换的开销，降低延迟。

source 和 算子map 如果是 one by one 的关系，他们的数据交换可以通过缓存而不是网络通信

TaskMar 为控制执行任务的数量，将计算资源划分多个slot,每个slot独享计算资源，这种静态分配利于任务资源隔离。

同一个任务可以共享一个slot, 不同作业不可以。

这里因为 Source 和 Map 并行度都是4 采用直连方式，他们的数据通信采用缓存形式

所以一共需要两个TaskMar source,Map 一个，reduce一个， 每个TaskMar 要3个slot

JobMar 将 JobGraph 部署 ExecutionGraph

设置的并行度，可以让一个ExecJobVertex 对应 多个并行的ExecVertex 实例。

Flink通过状态机管理 ExecGraph的作业执行进度。

Flink 将对象序列化为固定数量的预先分配的内存段，而不是直接把对象放在堆内存上。

Flink TaskMar 是由几个内部组件组成的：actor 系统（负责与 Flink 协调）、IOMar（负责将数据溢出到磁盘并将其读取回来）、MemoryMar（负责协调内存使用。

数据源：

Sink:

时间：

处理时间：取自Operator的机器系统时间

进入时间： 被Source观察时的系统时间

如果数据源没有自己正确创建水印，程序必须自己生成水印来确保基于的时间窗口可以正常工作。。

DataStream 提供了 周期性水印，间歇式水印，和递增式水印

flink中可以实现每n秒执行一个方法的定时任务吗？使用Ja自己的定时作是失效的

2：At-least-once或者Exectly-once消息投递模式是否有特殊要求

是的，Apache Flink可以实现每n秒执行一个方法的定时任务。Flink提供了两种实现方式：

Timer Serv是Flink中的一种内置机制，可以基于时间戳触发回调函数。在使用Timer Serv时，可以通过ProcessingTimeServ指定定时器的触发时间，例如每n秒触发一次。以下是使用Timer Serv实现每5秒执行一次方法的示例代码：

public class TimerExample {

public static void main(String[] args) throws Exception {

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

env.addSource(new SourceFunction() {

public void run(SourceContext ctx) throws Exception {

while (true) {

ctx.collect("Hello, World!");

Thread.sleep(5000);

public void cancel() {

}}).addSink(new SinkFunction() {

public void invoke(String value) throws Exception {

System.out.prin2：At-least-once或者Exectly-once消息投递模式是否有特殊要求tln(value);

}});

env.execute("Timer Example");

在上面的示例代码中，源函数中每5秒发出一个，Sink函数中接收并打印出。使用Timer Serv时，需要注意：

Timer Serv是基于时间戳的，因此需要确保中包含时间戳信息。

Timer Serv是由Flink系统驱动的，因此需要在Flink作业中执行。

2.使用Ja的ScheduledExecutorServ

除了Timer Serv之外，还可以使用Ja的ScheduledExecutorServ来实现定时任务。以下是使用ScheduledExecutorServ实现每5秒执行一次方法的示例代码：

public class ScheduledExecutorExample {

public static void main(String[] args) throws Exception {

ScheduledExecutorServ executorServ = Executors.newScheduledThreadPool(1);

executorServ.scheduleAtFixedRate(new Runnable() {

System.out.println("Hello, World!");

}}, 0, 5, TimeUnit.SECONDS);

Thread.sleep(60000);

executorServ.shutdown();

在上面的示例代码中，使用scheduleAtFixedRate方法指定每5秒执行一次方法。需要注意的是，使用ScheduledExecutorServ时，需要在Flink作业中执行。

import org.apache.flink.streaming.api.functions.source.SourceFunction;public class TimerTask {

public static void main(String[] args) throws Exception { StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

// 每秒执行一次 env.setStreamTimeCharacteristic(TimeCharacteristic.ProcessingTime); env.setParalleli(1);

env.addSource(new SourceFunction() {

public void run(SourceContext ctx) throws Exception { while (true) {

ctx.collect("Hello World");

Thread.sleep(1000);

} }

public void cancel() {

} }).print(); env.execute("TimerTask"); }}

Flink 原理详解

1.使用Flink的Timer Serv

Flink 是一个流处理框架，支持流处理和批处理，特点是流处理有限，可容错，可扩展，高吞吐，低延迟。

@Override

流处理是处理一条，立马下一个会从缓存中取出，在下一个进行计算

批处理是只有处理一批完成后，才会经过网络传输到下一个

流处理的优点是低延迟 批处理的优点是高吞吐

flink同时支持两种，flink的网络传输是设计固定的缓存块为单位，用户可以设置缓存块的超时值来决定换存块什么时候进行传输。 数据大于0 进行处理就是流式处理。

如果设置为无限大就是批处理模型。

Flink 集群包括 JobMar 和 TaskMar .

JobMar 主要负责调度 Job 并协调 Task 做 checkpoint，职责上很像 Storm 的 Nimbus。从 处接收到 Job 和 JAR 包 等资源后，会生成优化后的执行，并以 Task 的单元调度到各个 TaskMar 去执行。

TaskMar 在启动的时候就设置好了槽位数（Slot），每个 slot 能启动一个 Task，Task 为线程。从 JobMar 处接收需要 部署的 Task，部署启动后，与自己的上游建立 Netty 连接，接收数据并处理。

flink on yarn 是由client 提交 app到 RM 上， 然后RM 分配一个 AppMaster负责运行 Flink JobMar 和 Yarn AppMaster, 然后 AppMaster 分配 容器去运行 Flink TaskManger

Spark Streaming把实时输入数据流以时间片Δt （如1秒）为单位切分成块，Spark Streaming会把每块数据作为一个RDD，并使用RDD作处理每一小块数据。每个块都会生成一个Spark Job处理，然后分批次提交job到集群中去运行，运行每个 job的过程和真正的spark 任务没有任何区别。

JobScheduler, 负责 Job的调度通过定时器每隔一段时间根据Dstream的依赖关系生一个一个DAG图

ReceiverTracker负责数据的接收，管理和分配

ReceiverTracker在启动Receiver的时候他有ReceiverSupervisor,其实现是ReceiverSupervisorImpl, ReceiverSupervisor本身启 动的时候会启动Receiver，Receiver不断的接收数据，通过BlockGenerator将数据转换成Block。定时器会不断的把Block数据通会不断的把Block数据通过BlockMar或者WAL进行存储，数据存储之后ReceiverSupervisorlmpl会把存储后的数据的元数据Metadate汇报给ReceiverTracker，其实是汇报给ReceiverTracker中的RPC实体ReceiverTrackerEndpoin

spark on yarn 的cluster模式， Spark client 向RM提交job请求, RM会分配一个 AppMaster, driver 和 运行在AppMAster里， AM然后把Receiver作为一个Task提交给Spark Executor ， Receive启动接受数据，生成数据块，并通知Spark App, AM会根据数据块生成相应的Job, 并把Job 提交给空闲的 Executor 去执行。

1：需要关注流数据是否需要进行状态管理

3：对于小型的项目，并且需要低延迟的场景，建议使用storm

4：如果你的项目已经使用了spark，并且秒级别的实时处理可以满足需求的话，建议使用sparkStreaming

5：要求消息投递语义为 Exactly Once 的场景；数据量较大，要求高吞吐低延迟的场景；需要进行状态管理或窗口统计的场景，建议使用flink

Flink 提供的Api右 DataStream 和 DataSet ，他们都是不可变的数据，不可以增加删除中的元素， 通过 Source 创建 DataStream 和 DataSet

在创建运行时有：

Flink的每一个Operator称为一个任务， Operator 的每一个实例称为子任务，每一个任务在JVM线程中执行。可以将多个子任务链接成一个任务，减少上下文切换的开销，降低延迟。

source 和 算子map 如果是 one by one 的关系，他们的数据交换可以通过缓存而不是网络通信

TaskMar 为控制执行任务的数量，将计算资源划分多个slot,每个slot独享计算资源，这种静态分配利于任务资源隔离。

同一个任务可以共享一个slot, 不同作业不可以。

这里因为 Source 和 Map 并行度都是4 采用直连方式，他们的数据通信采用缓存形式

所以一共需要两个TaskMar source,Map 一个，reduce一个， 每个TaskMar 要3个slot

JobMar 将 JobGraph 部署 ExecutionGraph

设置的并行度，可以让一个ExecJobVertex 对应 多个并行的ExecVertex 实例。

Flink通过状态机管理 ExecGraph的作业执行进度。

Flink 将对象序列化为固定数量的预先分配的内存段，而不是直接把对象放在堆内存上。

Flink TaskMar 是由几个内部组件组成的：actor 系统（负责与 Flink 协调）、IOMar（负责将数据溢出到磁盘并将其读取回来）、MemoryMar（负责协调内存使用。

数据源：

Sink:

时间：

处理时间：取自Operator的机器系统时间

进入时间： 被Source观察时的系统时间

如果数据源没有自己正确创建水印，程序必须自己生成水印来确保基于的时间窗口可以正常工作。。

DataStream 提供了 周期性水印，间歇式水印，和递增式水印