Shulou(Shulou.com)06/01 Report--

This article mainly explains "how to achieve Flink timer". The content in the article is simple and clear, and it is easy to learn and understand. Please follow the editor's train of thought to study and learn how to realize Flink timer.

Background demand

In the e-commerce field, there will be such a scenario, if the user buys a product and does not make an evaluation within 24 hours after the order is completed, the system will automatically give five-star praise. Today, we mainly use flink timer to simply achieve this function.

Case study on Custom source

First of all, we still use custom source to simulate and generate some order data.

Here, we give birth to the simplest tuple Tuple2, which contains two fields: order id and order completion time.

Public static class MySource implements SourceFunction {

Private volatile boolean isRunning = true

@ Override

Public void run (SourceContext ctx) throws Exception {

While (isRunning) {

Thread.sleep (1000)

/ / order id

String orderid = UUID.randomUUID () .toString ()

/ / order completion time

Long orderFinishTime = System.currentTimeMillis ()

Ctx.collect (Tuple2.of (orderid, orderFinishTime))

}

}

@ Override

Public void cancel () {

IsRunning = false

}

}

Timing processing logic

Let's code first, and then we'll explain the code in turn.

Public static class TimerProcessFuntion

Extends KeyedProcessFunction {

Private MapState mapState

/ / for how long (interval, milliseconds) there is no evaluation, then automatic five-star praise

Private long interval = 0l

Public TimerProcessFuntion (long interval) {

This.interval = interval

}

@ Override

Public void open (Configuration parameters) {

MapStateDescriptor mapStateDesc = new MapStateDescriptor (

"mapStateDesc"

String.class, Long.class)

MapState = getRuntimeContext () .getMapState (mapStateDesc)

}

@ Override

Public void onTimer (

Long timestamp, OnTimerContext ctx, Collector out) throws Exception {

Iterator iterator = mapState.iterator ()

While (iterator.hasNext ()) {

Map.Entry entry = (Map.Entry) iterator.next ()

String orderid = entry.getKey ()

Boolean f = isEvaluation (entry.getKey ())

MapState.remove (orderid)

If (f) {

LOG.info ("order (orderid: {}) has been evaluated and not processed within {} milliseconds", orderid, interval)

}

If (f) {

/ / if the user does not make an evaluation, the default five-star evaluation is given when calling the relevant API.

LOG.info ("order (orderid: {}) is not evaluated for more than {} milliseconds, call API to give five-star automatic praise", orderid, interval)

}

}

}

/ * *

* whether the user has evaluated the order, in the production environment, you can query the relevant order system.

* We just made a random judgment here.

*

* @ param key

* @ return

, /

Private boolean isEvaluation (String key) {

Return key.hashCode ()% 2 = 0

}

@ Override

Public void processElement (

Tuple2 value, Context ctx, Collector out) throws Exception {

MapState.put (value.f0, value.f1)

Ctx.timerService () .registerProcessingTimeTimer (value.f1 + interval)

}

}

First of all, we define a state of type MapState. Key is the order number, and value is the order completion time. When processElement processes data, the information of each order is stored in the status. No processing is done at this time, and register a timer that is longer than the order completion time (interval). The registered timing task triggers the onTimer method when it reaches the timer, which we mainly deal with. We call the external interface to determine whether the user has made an evaluation. If there is no evaluation, call the API to give five-star praise, and if we do, we will not deal with anything. Finally, remember to delete the corresponding order from MapState and thank you for your reading. This is the content of "how to implement Flink timer". After the study of this article, I believe you have a deeper understanding of how to realize Flink timer. The specific use situation still needs to be verified by practice. Here is, the editor will push for you more related knowledge points of the article, welcome to follow!

Welcome to subscribe "Shulou Technology Information " to get latest news, interesting things and hot topics in the IT industry, and controls the hottest and latest Internet news, technology news and IT industry trends.