上接 链接
package day5;
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.serialization.SimpleStringSchema;
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.tuple.Tuple3;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
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.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import java.text.SimpleDateFormat;
import java.time.Duration;
import java.util.Date;
import java.util.Properties;
public class Std9Timer {
public static void main(String[] args) throws Exception {
final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(1);
// 定义kafka相关配置项
Properties pro = new Properties();
pro.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,"hadoop1:9092,hadoop2:9092,hadoop3:9092");
pro.setProperty(ConsumerConfig.GROUP_ID_CONFIG,"consumer-group");
pro.setProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG,"latest");
// 内置连接器，创建FlinkKafka消费者，需要指定主题、反序列化、kafka配置
FlinkKafkaConsumer<String> fkc = new FlinkKafkaConsumer<>("temp", new SimpleStringSchema(), pro);
DataStreamSource<String> text = env.addSource(fkc);
WatermarkStrategy<Tuple3<String, String, Long>> ws = WatermarkStrategy
// 创建 水位线对象，并指定延迟时间。该方法用于处理乱序数据
.<Tuple3<String, String, Long>>forBoundedOutOfOrderness(Duration.ofSeconds(0))
// 指定 数据 中的 时间戳
.withTimestampAssigner(new SerializableTimestampAssigner<Tuple3<String, String, Long>>() {
@Override
public long extractTimestamp(Tuple3<String, String, Long> element, long recordTimestamp) {
return element.f2;
}
});
// 定义侧流
OutputTag ot = new OutputTag("orderTimeout", TypeInformation.of(OrderResult.class));
// 转换操作
SingleOutputStreamOperator<OrderResult> soso = text
.flatMap(new FlatMapFunImpl())
.assignTimestampsAndWatermarks(ws)// 设置水位线
.keyBy(value -> value.f0) // 根据tuple的第一个元素（温度传感器的id）进行分组
.process(new MyKeyedProcessFunction(ot));// 侧流传入 ProcessFunction，将无效的订单的结果放入到该流中
soso.print("主流--");
soso.getSideOutput(ot).print("侧流--");
System.out.println(env.getExecutionPlan());
env.execute("myjob");
}
// kafka 发送的数据格式是 json，我们接收的是普通字符串，需要转换成 json 对象
static class FlatMapFunImpl implements FlatMapFunction<String,Tuple3<String,String, Long>>{
@Override
public void flatMap(String value, Collector<Tuple3<String,String, Long>> out) throws Exception {
// 样例数据：100,create,2019-05-21 17:27:22
String[] arr = value.split(",");
SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
Date date = sdf.parse(arr[2]);
out.collect(Tuple3.of(arr[0],arr[1],date.getTime()));
Thread.sleep(200);
}
}
static class MyKeyedProcessFunction extends KeyedProcessFunction<String,Tuple3<String,String, Long>,OrderResult>{
/**
* 延时10分钟，由于定时单位是：毫秒，所以需要转换
*/
// public static final int DELAY = 10 * 60 * 1000;
/**
* 延时10秒，由于定时单位是：毫秒，所以需要转换
*/
public static final int DELAY = 10 * 1000;
// 状态编程，存储 定时器 时间。因为在 taskmanager中运行，如果是普通成员变量，存不住数据
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(); // 调用 update 更新的时间（延迟10分钟）
SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
System.out.println("接收数据::"+value+",事件时间："+sdf.format(value.f2));
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+",调用 onTimer() 时间："+timer);
// 设置定时器，单位：毫秒
ctx.timerService().registerProcessingTimeTimer(ts);
timerState.update(ts);
} else if ("pay".equals(value.f1)) {
// 接收到支付订单
if(timerTs == null ){
// 执行定时onTimer() 方法后，timerTs为null，说明 该数据 已经超时了，本次支付无效
ctx.output(orderTimeoutOT,new OrderResult(value.f0, "订单已超时，本次支付无效"));
}else {
// 还没有执行定时onTimer() 方法，就接收到 支付数据了，那么就删除定时器
// 输出数据
out.collect(new OrderResult(value.f0, "支付成功"));
// 删除定时器
ctx.timerService().deleteProcessingTimeTimer(timerTs);
System.out.println(value.f0 + "删除定时器");
}
}
}
// 定时器到时要执行的操作
@Override
public void onTimer(long timestamp, OnTimerContext ctx, Collector<OrderResult> out) throws Exception {
// 将超时订单放入到侧流中
ctx.output(orderTimeoutOT,new OrderResult(ctx.getCurrentKey(), "订单已超时，自动取消"));
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();
}
}
static class OrderResult{
String id;
String info;
public OrderResult(String id, String info) {
this.id = id;
this.info = info;
}
@Override
public String toString() {
return "OrderResult{" +
"id='" + id + '\'' +
", info='" + info + '\'' +
'}';
}
}
}
测试类
第一列：订单id
第二列：订单状态，create 下订单，pay 支付
第三列：表示当前时间延后的秒数
100,create,0
101,create,1
100,pay,2
102,create,4
103,create,10
104,create,13
103,pay,53
105,create,63
104,pay,664
105,pay,1260
启动类
package day5;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import util.TextFileReader;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.List;
import java.util.Properties;
public class 订单超时 {
final static String TOPIC = "temp";
public static void main(String[] args) throws Exception {
// 1. 创建 kafka 生产者的配置对象
Properties properties = new Properties();
// 2. 给 kafka 配置对象添加配置信息：bootstrap.servers
properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"hadoop1:9092,hadoop2:9092,hadoop3:9092");
// key,value 序列化（必须）：key.serializer，value.serializer
properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,
"org.apache.kafka.common.serialization.StringSerializer");
// StringSerializer.class.getName() 等于 org.apache.kafka.common.serialization.StringSerializer
properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,
"org.apache.kafka.common.serialization.StringSerializer") ;
// 3. 创建 kafka 生产者对象
KafkaProducer<String, String> kafkaProducer = new KafkaProducer(properties);
List<String> list = new TextFileReader("./data/订单.txt").readList();
Date date = new Date();
SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
long timestamp = date.getTime();
for(String line : list) {
String[] arr = line.split(",");
Integer t = Integer.parseInt(arr[2])*1000;
Date temp = new Date(timestamp+t);
String time = sdf.format(temp);
String s = arr[0]+","+arr[1]+","+time;
// 4. 调用 send 方法,发送消息，全部延迟2秒
kafkaProducer.send(new ProducerRecord(TOPIC , s));
System.out.println("send:" + s);
Thread.sleep(500);
}
// 5. 关闭资源
kafkaProducer.close();
System.out.println("end...");
}
}