flink實戰 —— 定時器實作已完成訂單自動五星好評

雲栖号資訊：【 點選檢視更多行業資訊

】

在這裡您可以找到不同行業的第一手的上雲資訊，還在等什麼，快來！

背景需求

在電商領域會有這麼一個場景，如果使用者買了商品，在訂單完成之後，24小時之内沒有做出評價，系統自動給與五星好評，我們今天主要使用flink的定時器來簡單實作這一功能。

案例詳解

自定義source

首先我們還是通過自定義source來模拟生成一些訂單資料。

在這裡，我們生了一個最簡單的二進制組Tuple2,包含訂單id和訂單完成時間兩個字段。

public static class MySource implements SourceFunction<Tuple2<String,Long>>{
  private volatile boolean isRunning = true;

  @Override
  public void run(SourceContext<Tuple2<String,Long>> ctx) throws Exception{
   while (isRunning){
    Thread.sleep(1000);
    //訂單id
    String orderid = UUID.randomUUID().toString();
    //訂單完成時間
    long orderFinishTime = System.currentTimeMillis();
    ctx.collect(Tuple2.of(orderid, orderFinishTime));
   }
  }

  @Override
  public void cancel(){
   isRunning = false;
  }
 }           

定時處理邏輯

先上代碼, 我們再來依次解釋代碼

public static class TimerProcessFuntion
   extends KeyedProcessFunction<Tuple,Tuple2<String,Long>,Object>{

  private MapState<String,Long> mapState;
  //超過多長時間(interval,機關：毫秒) 沒有評價，則自動五星好評
  private long interval = 0l;

  public TimerProcessFuntion(long interval){
   this.interval = interval;
  }

  @Override
  public void open(Configuration parameters){
   MapStateDescriptor<String,Long> mapStateDesc = new MapStateDescriptor<>(
     "mapStateDesc",
     String.class, Long.class);
   mapState = getRuntimeContext().getMapState(mapStateDesc);
  }

  @Override
  public void onTimer(
    long timestamp, OnTimerContext ctx, Collector<Object> out) throws Exception{
   Iterator iterator = mapState.iterator();
   while (iterator.hasNext()){
    Map.Entry<String,Long> entry = (Map.Entry<String,Long>) iterator.next();

    String orderid = entry.getKey();
    boolean f = isEvaluation(entry.getKey());
    mapState.remove(orderid);
    if (f){
     LOG.info("訂單(orderid: {}) 在  {} 毫秒時間内已經評價，不做處理", orderid, interval);
    }
    if (f){
     //如果使用者沒有做評價，在調用相關的接口給與預設的五星評價
     LOG.info("訂單(orderid: {}) 超過  {} 毫秒未評價，調用接口給與五星自動好評", orderid, interval);
    }
   }
  }

  /**
   * 使用者是否對該訂單進行了評價，在生産環境下，可以去查詢相關的訂單系統.
   * 我們這裡隻是随便做了一個判斷
   *
   * @param key
   * @return
   */
  private boolean isEvaluation(String key){
   return key.hashCode() % 2 == 0;
  }

  @Override
  public void processElement(
    Tuple2<String,Long> value, Context ctx, Collector<Object> out) throws Exception{
   mapState.put(value.f0, value.f1);
   ctx.timerService().registerProcessingTimeTimer(value.f1 + interval);
  }
 }           

• 首先我們定義一個MapState類型的狀态，key是訂單号，value是訂單完成時間。
• 在processElement處理資料的時候，把每個訂單的資訊存入狀态中，這個時候不做任何處理，并且注冊一個比訂單完成時間大于間隔時間(interval)的定時器.
• 注冊的定時任務在到達了定時器的時間就會觸發onTimer方法，我們主要在這個裡面進行處理。我們調用外部的接口來判斷使用者是否做過評價，如果沒做評價，調用接口給與五星好評，如果做過評價，則什麼也不處理,最後記得把相應的訂單從MapState删除。

【雲栖号線上課堂】每天都有産品技術專家分享！

課程位址：

https://yqh.aliyun.com/live

立即加入社群，與專家面對面，及時了解課程最新動态！

【雲栖号線上課堂 社群】

https://c.tb.cn/F3.Z8gvnK

原文釋出時間：2020-06-16

本文作者：大資料技術與應用實戰

本文來自：“

掘金

”，了解相關資訊可以關注“掘金”