在上一篇StreamOperator源碼簡析從源碼角度分析了StreamOperator以及其實現(xiàn)類，此篇幅主要分析一下如何自定義一個StreamOperator仿荆。

StreamOperator接口提供了其生命周期的抽象方法，例如初始化方法setup坏平、open拢操、initializeState，checkpoint相關(guān)方法prepareSnapshotPreBarrier舶替、snapshotState令境，但是我們沒有必要去自己一一實現(xiàn)這些方法，可以繼承其抽象類AbstractStreamOperator顾瞪，覆蓋一些我們需要重寫的方法舔庶。在上一篇分析中提到對于source端不需要接受上游數(shù)據(jù)抛蚁，也就不需要實現(xiàn)OneInputStreamOperator或者TwoInputStreamOperator接口，如果我們需要接收上游數(shù)據(jù)就必須實現(xiàn)這兩個接口中的一個惕橙，主要看一個輸入還是兩個輸入來選擇瞧甩。
案例：假設(shè)我們現(xiàn)在需要實現(xiàn)一個通用的定時、定量的輸出的StreamOperator吕漂。
實現(xiàn)步驟：

1. 繼承AbstractStreamOperator抽象類亲配，實現(xiàn)OneInputStreamOperator接口

2. 重寫open方法尘应，調(diào)用flink 提供的定時接口惶凝，并且注冊定時器

3. 重寫initializeState/snapshotState方法，由于批量寫需要做緩存犬钢，那么需要保證數(shù)據(jù)的一致性苍鲜，將緩存數(shù)據(jù)存在狀態(tài)中

4. 重寫processElement方法，將數(shù)據(jù)存在緩存中玷犹，達(dá)到一定大小然后輸出

5. 由于需要做定時調(diào)用混滔，那么需要有一個定時調(diào)用的回調(diào)方法，那么定義的類需要實現(xiàn)ProcessingTimeCallback接口歹颓，并且實現(xiàn)其onProcessingTime方法（關(guān)于flink定時可以參考定時系列文章）

代碼：

1. publicabstractclassCommonSinkOperator<T extendsSerializable>extendsAbstractStreamOperator<Object>

2. implementsProcessingTimeCallback,OneInputStreamOperator<T,Object>{

3. privateList<T> list;

4. privateListState<T> listState;

5. privateint batchSize;

6. privatelong interval;

7. privateProcessingTimeService processingTimeService;

8. publicCommonSinkOperator(){

9. }

10. publicCommonSinkOperator(int batchSize,long interval){

11. this.chainingStrategy =ChainingStrategy.ALWAYS;

12. this.batchSize = batchSize;

13. this.interval = interval;

14. }

15. @Overridepublicvoid open()throwsException{

16. super.open();

17. if(interval >0&& batchSize >1){

18. //獲取AbstractStreamOperator里面的ProcessingTimeService坯屿， 該對象用來做定時調(diào)用

19. //注冊定時器將當(dāng)前對象作為回調(diào)對象，需要實現(xiàn)ProcessingTimeCallback接口

20. processingTimeService = getProcessingTimeService();

21. long now = processingTimeService.getCurrentProcessingTime();

22. processingTimeService.registerTimer(now + interval,this);

23. }

24. }

25. //狀態(tài)恢復(fù)

26. @Overridepublicvoid initializeState(StateInitializationContext context)throwsException{

27. super.initializeState(context);

28. this.list =newArrayList<T>();

29. listState = context.getOperatorStateStore().getSerializableListState("batch-interval-sink");

30. if(context.isRestored()){

31. listState.get().forEach(x ->{

32. list.add(x);

33. });

34. }

35. }

36. @Overridepublicvoid processElement(StreamRecord<T> element)throwsException{

37. list.add(element.getValue());

38. if(list.size()>= batchSize){

39. saveRecords(list);

40. }

41. }

42. //checkpoint

43. @Overridepublicvoid snapshotState(StateSnapshotContext context)throwsException{

44. super.snapshotState(context);

45. if(list.size()>0){

46. listState.clear();

47. listState.addAll(list);

48. }

49. }

50. //定時回調(diào)

51. @Overridepublicvoid onProcessingTime(long timestamp)throwsException{

52. if(list.size()>0){

53. saveRecords(list);

54. list.clear();

55. }

56. long now = processingTimeService.getCurrentProcessingTime();

57. processingTimeService.registerTimer(now + interval,this);//再次注冊

58. }

59. publicabstractvoid saveRecords(List<T> datas);

60. }

如何調(diào)用巍扛？直接使用dataStream.transform方式即可领跛。

整體來說這個demo相對來說是比較簡單的，但是這里面涉及的定時撤奸、狀態(tài)管理也是值得研究吠昭，比喻說在這里定時我們直接選擇ProcessingTimeService，而沒有選擇InternalTimerService來完成定時注冊胧瓜，主要是由于InternalTimerService會做定時調(diào)用狀態(tài)保存矢棚，在窗口操作中需要任務(wù)失敗重啟仍然可以觸發(fā)定時，但是在我們案例中不需要府喳，直接下次啟動重新注冊即可蒲肋，因此選擇了ProcessingTimeService。

推薦閱讀

1. Flink中延時調(diào)用設(shè)計與實現(xiàn)

2. Flink維表關(guān)聯(lián)系列之Hbase維表關(guān)聯(lián)：LRU策略

3. 你應(yīng)該了解的Watermark

4. Flink exactly-once系列之事務(wù)性輸出實現(xiàn)

5. Flink時間系統(tǒng)系列之實例講解：如何做定時輸出

6. Flink實戰(zhàn):全局TopN分析與實現(xiàn)

7. Flink per-Job模式InfluxdbReporter上報JobName

8. Flink SQL自定義聚合函數(shù)

image

關(guān)注回復(fù)Flink獲取更多信息~

image