treamOperator接口提供了其生命周期的抽象方法，例如初始化方法setup、open、initializeState，checkpoint相关方法prepareSnapshotPreBarrier、snapshotState，但是我们没有必要去自己一一实现这些方法，可以继承其抽象类AbstractStreamOperator，覆盖一些我们需要重写的方法。在上一篇分析中提到对于source端不需要接受上游数据，也就不需要实现OneInputStreamOperator或者TwoInputStreamOperator接口，如果我们需要接收上游数据就必须实现这两个接口中的一个，主要看一个输入还是两个输入来选择。
案例：假设我们现在需要实现一个通用的定时、定量的输出的StreamOperator。
实现步骤：
继承AbstractStreamOperator抽象类，实现OneInputStreamOperator接口
重写open方法，调用flink 提供的定时接口，并且注册定时器
重写initializeState/snapshotState方法，由于批量写需要做缓存，那么需要保证数据的一致性，将缓存数据存在状态中
重写processElement方法，将数据存在缓存中，达到一定大小然后输出
由于需要做定时调用，那么需要有一个定时调用的回调方法，那么定义的类需要实现ProcessingTimeCallback接口，并且实现其onProcessingTime方法（关于flink定时可以参考定时系列文章）
代码：
publicabstractclassCommonSinkOperator<T extendsSerializable>extendsAbstractStreamOperator<Object>implementsProcessingTimeCallback,OneInputStreamOperator<T,Object>{privateList<T> list;privateListState<T> listState;privateint batchSize;privatelong interval;privateProcessingTimeService processingTimeService;publicCommonSinkOperator(){}publicCommonSinkOperator(int batchSize,long interval){this.chainingStrategy =ChainingStrategy.ALWAYS;this.batchSize = batchSize;this.interval = interval;}@Overridepublicvoid open()throwsException{super.open();if(interval >0&& batchSize >1){//获取AbstractStreamOperator里面的ProcessingTimeService， 该对象用来做定时调用//注册定时器将当前对象作为回调对象，需要实现ProcessingTimeCallback接口processingTimeService = getProcessingTimeService();long now = processingTimeService.getCurrentProcessingTime();processingTimeService.registerTimer(now + interval,this);}}//状态恢复@Overridepublicvoid initializeState(StateInitializationContext context)throwsException{super.initializeState(context);this.list =newArrayList<T>();listState = context.getOperatorStateStore().getSerializableListState("batch-interval-sink");if(context.isRestored()){listState.get().forEach(x ->{list.add(x);});}}@Overridepublicvoid processElement(StreamRecord<T> element)throwsException{list.add(element.getValue());if(list.size()>= batchSize){saveRecords(list);}}//checkpoint@Overridepublicvoid snapshotState(StateSnapshotContext context)throwsException{super.snapshotState(context);if(list.size()>0){listState.clear();listState.addAll(list);}}//定时回调@Overridepublicvoid onProcessingTime(long timestamp)throwsException{if(list.size()>0){saveRecords(list);list.clear();}long now = processingTimeService.getCurrentProcessingTime();processingTimeService.registerTimer(now + interval,this);//再次注册}publicabstractvoid saveRecords(List<T> datas);}
如何调用？直接使用dataStream.transform方式即可。
整体来说这个demo相对来说是比较简单的，但是这里面涉及的定时、状态管理也是值得研究，比喻说在这里定时我们直接选择ProcessingTimeService，而没有选择InternalTimerService来完成定时注册，主要是由于InternalTimerService会做定时调用状态保存，在窗口操作中需要任务失败重启仍然可以触发定时，但是在我们案例中不需要，直接下次启动重新注册即可，因此选择了ProcessingTimeService。