说明
功能
可以实现 聚合统计，设置 定时器，还可以访问 watermark 等
使用
先用 keyBy() 算子对数据流进行处理，得到一个 KeyedStream，然后调用 process() 方法
也就是指定一个键（key），按照它的哈希值（hash code）将数据分成不同的“组”，然后分配到不同的并行子任务上执行计算
TimerService
通过 TimerService 可以访问 时间 和定时器，包含以下方法：
// 获取当前的处理时间
long currentProcessingTime();
// 获取当前的水位线（事件时间）
long currentWatermark();
// 注册处理时间定时器，当处理时间超过 time 时触发
void registerProcessingTimeTimer(long time);
// 注册事件时间定时器，当水位线超过 time 时触发
void registerEventTimeTimer(long time);
// 删除触发时间为 time 的处理时间定时器
void deleteProcessingTimeTimer(long time);
// 删除触发时间为 time 的处理时间定时器
void deleteEventTimeTimer(long time);
获取 TimerService 对象
通过下面方法获取 TimerService
public void processElement(T value, Context ctx, Collector<> out) throws Exception{
TimerService ts = ctx.timerService()
}
定时
设置定时
ctx.timerService().registerEventTimeTimer(time);
参数：
time：是事件时间 + n 毫秒，表示 n 毫秒后定时启动
定时方法
重写下面方法，定时启动，就会执行该方法
@Override
public void onTimer(long timestamp, OnTimerContext ctx, Collector<> out) throws Exception {
}
触发定时方法的时机（关键）
当
watermark大于等于定时时间时，才会执行 定时方法onTimer()没有数据时，会触发所有的定时器
注意：这里的定时触发，与闹钟不同，闹钟是：定时5分钟后闹铃响。而 flink 基于事件时间的定时，与 watermark 和 事件时间 相关
案例
废弃，新案例见 链接
在电商平台中，下单后，拖延一段时间后，用户支付的意愿会降低。
所以为了让用户更有紧迫感从而提高支付转化率，同时也为了防范订单支付环节的安全风险，电商网站往往会对订单状态进行监控，设置一个失效时间，如：10 分钟，如果下单后 10 分钟内未支付，订单就会被取消。
准备数据
100,create,2019-05-21 17:27:22
101,create,2019-05-21 17:27:23
100,pay,2019-05-21 17:27:24
102,create,2019-05-21 17:27:26
103,create,2019-05-21 17:27:32
104,create,2019-05-21 17:27:35
103,pay,2019-05-21 17:28:15
105,create,2019-05-21 17:28:25
104,pay,2019-05-21 17:38:26
105,pay,2019-05-21 17:48:00
列解释：
- 第一列：订单id
- 第二列：订单状态，
create表示创建订单，pay表示支付订单 - 第三列：事件时间
数据说明：
101、102订单没有支付，所以 flink 程序应识别出来104、105支付时间超过10分钟，支付无效，flink程序应识别出来
java
为了简单清晰，不考虑乱序（否则有bug）
package processfunction;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;
import java.text.SimpleDateFormat;
import java.time.Duration;
import java.util.Date;
/**
* 不考虑乱序
*/
public class StdProcessFunction定时 {
public static void main(String[] args) throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(1);
DataStream<String> ds = env.readTextFile("data/3.csv");
SingleOutputStreamOperator<Tuple3<String, String, Long>> soso = ds.flatMap(new FlatMapFunction<String, Tuple3<String, String, Long>>() {
@Override
public void flatMap(String value, Collector<Tuple3<String,String, Long>> out) throws Exception {
String[] arr = value.split(",");
// 时间戳单位：毫秒，文件中的时间是10个数字，单位是秒，所以需要乘1000
SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
Date date = sdf.parse(arr[2]);
Tuple3 data = Tuple3.of(arr[0],arr[1],date.getTime());
out.collect(data);
Thread.sleep(300);
}
});
SingleOutputStreamOperator<Tuple3<String, String, Long>> watermarksSOSO = soso
.assignTimestampsAndWatermarks(
WatermarkStrategy.<Tuple3<String, String, Long>>forBoundedOutOfOrderness(Duration.ZERO)
.withTimestampAssigner(new SerializableTimestampAssigner<Tuple3<String, String, Long>>() {
@Override
public long extractTimestamp(Tuple3<String, String, Long> element, long recordTimestamp) {
return element.f2;
}
})
);
KeyedStream<Tuple3<String, String, Long>, String> keyedStream = watermarksSOSO.keyBy(new KeySelector<Tuple3<String, String, Long>, String>() {
@Override
public String getKey(Tuple3<String, String, Long> value) throws Exception {
return value.f0;
}
});
// 侧流
OutputTag<OrderResult> orderTimeoutOT = new OutputTag<>("orderTimeout", TypeInformation.of(OrderResult.class));
SingleOutputStreamOperator<OrderResult> process = keyedStream.process(new MyKeyedProcessFunction(orderTimeoutOT));
// 获取侧流
DataStream<OrderResult> otDS = process.getSideOutput(orderTimeoutOT);
// 这里为了控制台打印消息清晰，所以没有打印
// process.print("--");
// otDS.print("==");
env.execute();
}
static class MyKeyedProcessFunction extends KeyedProcessFunction<String, Tuple3<String, String, Long>, OrderResult> {
/**
* 延时10分钟，由于定时单位是：毫秒，所以需要转换
*/
public static final int DELAY = 10 * 60 * 1000;
/**
* 延时5秒，由于定时单位是：毫秒，所以需要转换
*/
// public static final int DELAY = 5 * 1000;
private ValueState<Long> timerState ;
private OutputTag<OrderResult> orderTimeoutOT;
public MyKeyedProcessFunction(OutputTag<OrderResult> orderTimeoutOT) {
this.orderTimeoutOT = orderTimeoutOT;
}
@Override
public void open(Configuration parameters) throws Exception {
timerState = getRuntimeContext().getState(new ValueStateDescriptor<Long>("timer-state", Long.class));
}
@Override
public void processElement(Tuple3<String, String, Long> value, Context ctx, Collector<OrderResult> out) throws Exception{
Long timerTs = timerState.value();
SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
String watermark = sdf.format(new Date(ctx.timerService().currentWatermark()));
System.out.println("接收数据::"+value+",事件时间："+sdf.format(value.f2)+",Watermark:"+watermark);
if ("create".equals(value.f1)) {
// 接收到创建订单,设置定时器
// 计算定时时间，在事件时间之后的 n毫秒，单位：毫秒
long ts = value.f2 + DELAY;
// 格式化时间，方便打印查看
String timer = sdf.format(ts);
String eventTimeHuman = sdf.format(value.f2);
System.out.println("定时信息::"+value.f0+",事件时间："+eventTimeHuman+",定时时间："+timer);
// 设置定时器，单位：毫秒
ctx.timerService().registerEventTimeTimer(ts);
timerState.update(ts);
} else if ("pay".equals(value.f1)) {
// 接收到支付订单
if(timerTs == null){
// 执行定时onTimer() 方法后，timerTs为null，说明已经超时了，本次支付无效
out.collect(new OrderResult(value.f0, "订单已超时，本次支付无效"));
System.out.println(value.f0 + "订单已超时，本次支付无效");
}else {
// 还没有执行定时onTimer() 方法，就接收到 支付数据了，那么就删除定时器
// 输出数据
out.collect(new OrderResult(value.f0, "支付成功"));
System.out.println(value.f0 + "支付成功，删除响应的定时器");
// 删除定时器
ctx.timerService().deleteEventTimeTimer(timerTs);
}
timerState.clear();
}
}
// 定时器到时要执行的操作
@Override
public void onTimer(long timestamp, OnTimerContext ctx, Collector<OrderResult> out) throws Exception {
// 将超时订单放入到侧流中
ctx.output(orderTimeoutOT,new OrderResult(ctx.getCurrentKey(), " order timeout"));
SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
String tsHumer = sdf.format(timestamp);
System.out.println(String.format("==执行定时方法::%s 订单超时,timestamp:%s",ctx.getCurrentKey(),tsHumer));
// 执行定时方法后，将状态清空
timerState.clear();
}
}
}
执行结果
接收数据::(100,create,1558430842000),事件时间：2019-05-21 17:27:22,Watermark:292269055-12-03 00:47:04
定时信息::100,事件时间：2019-05-21 17:27:22,定时时间：2019-05-21 17:42:22
接收数据::(101,create,1558430843000),事件时间：2019-05-21 17:27:23,Watermark:2019-05-21 17:27:21
定时信息::101,事件时间：2019-05-21 17:27:23,定时时间：2019-05-21 17:42:23
接收数据::(100,pay,1558430844000),事件时间：2019-05-21 17:27:24,Watermark:2019-05-21 17:27:22
100支付成功，删除响应的定时器
接收数据::(102,create,1558430846000),事件时间：2019-05-21 17:27:26,Watermark:2019-05-21 17:27:23
定时信息::102,事件时间：2019-05-21 17:27:26,定时时间：2019-05-21 17:42:26
接收数据::(103,create,1558430852000),事件时间：2019-05-21 17:27:32,Watermark:2019-05-21 17:27:25
定时信息::103,事件时间：2019-05-21 17:27:32,定时时间：2019-05-21 17:42:32
接收数据::(104,create,1558430855000),事件时间：2019-05-21 17:27:35,Watermark:2019-05-21 17:27:31
定时信息::104,事件时间：2019-05-21 17:27:35,定时时间：2019-05-21 17:42:35
接收数据::(103,pay,1558430895000),事件时间：2019-05-21 17:28:15,Watermark:2019-05-21 17:27:34
103支付成功，删除响应的定时器
接收数据::(105,create,1558430905000),事件时间：2019-05-21 17:28:25,Watermark:2019-05-21 17:28:14
定时信息::105,事件时间：2019-05-21 17:28:25,定时时间：2019-05-21 17:43:25
接收数据::(104,pay,1558432035000),事件时间：2019-05-21 17:47:15,Watermark:2019-05-21 17:28:24
104支付成功，删除响应的定时器
==执行定时方法::101 订单超时,timestamp:2019-05-21 17:42:23
==执行定时方法::102 订单超时,timestamp:2019-05-21 17:42:26
==执行定时方法::105 订单超时,timestamp:2019-05-21 17:43:25
接收数据::(105,pay,1558432080000),事件时间：2019-05-21 17:48:00,Watermark:2019-05-21 17:47:14
105订单已超时，本次支付无效
解释：
bug
104 支付数据已经超过 10分钟了，但仍然支付成功，因为没有 watermark 大于 该数据的定时时间，所以定时器没有被触发
参考：
https://cloud.tencent.com/developer/article/1593443
https://blog.csdn.net/weixin_47491957/article/details/124451027