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);
  }
 }


           

複制

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

完整的代碼請參考

https://github.com/zhangjun0x01/bigdata-examples/blob/master/flink/src/main/java/timer/AutoEvaluation.java