Flink timer定时器

这篇具有很好参考价值的文章主要介绍了Flink timer定时器。希望对大家有所帮助。如果存在错误或未考虑完全的地方,请大家不吝赐教,您也可以点击"举报违法"按钮提交疑问。

Flink timer定时器

常见timer

基于处理时间或者事件时间处理过一个元素之后, 注册一个定时器, 然后指定的时间执行.

Context和OnTimerContext 所持有的TimerService对象拥有以下方法:

currentProcessingTime(): Long 返回当前处理时间

currentWatermark(): Long 返回当前watermark的时间戳

registerProcessingTimeTimer(timestamp: Long): Unit会注册当前key的processing time的定时器。当processing time到达定时时间时,触发timer。

registerEventTimeTimer(timestamp: Long): Unit 会注册当前key的event time 定时器。当水位线大于等于定时器注册的时间时,触发定时器执行回调函数。

deleteProcessingTimeTimer(timestamp: Long): Unit 删除之前注册处理时间定时器。如果没有这个时间戳的定时器,则不执行。

deleteEventTimeTimer(timestamp: Long): Unit 删除之前注册的事件时间定时器,如果没有此时间戳的定时器,则不执行。

案例一:基于wordcount 的定时器
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

public class ProcessTimerDemo {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
        //1000,hello
        DataStreamSource<String> lines = env.socketTextStream("localhost", 8888);
        SingleOutputStreamOperator<Tuple2<String, Integer>> wordAndCount = lines.map(new MapFunction<String, Tuple2<String, Integer>>() {
            @Override
            public Tuple2<String, Integer> map(String line) throws Exception {
                String[] fields = line.split(",");
                return Tuple2.of(fields[0], Integer.parseInt(fields[1]));
            }
        });
        KeyedStream<Tuple2<String, Integer>, String> keyed = wordAndCount.keyBy(tp -> tp.f0);

        SingleOutputStreamOperator<Tuple2<String, Integer>> process = keyed.process(new KeyedProcessFunction<String, Tuple2<String, Integer>, Tuple2<String, Integer>>() {
            private transient ValueState<Integer> counter;

            @Override
            public void open(Configuration parameters) throws Exception {
                ValueStateDescriptor<Integer> stateDescriptor = new ValueStateDescriptor<>("wc-state", Integer.class);
                counter = getRuntimeContext().getState(stateDescriptor);

            }

            @Override
            public void processElement(Tuple2<String, Integer> value, Context ctx, Collector<Tuple2<String, Integer>> out) throws Exception {
                // 获取当前的ProcessingTime
                long currentProcessingTime = ctx.timerService().currentProcessingTime();

                //注册定时器,如果注册的是ProcessingTime类型的定时器,当SubTask所在机器的ProcessingTime大于等于注册定时器的时间,就会触发onTimer方法
                long fireTime = currentProcessingTime - currentProcessingTime % 60000 + 60000;
                //注册ProcessingTime的定时器
                ctx.timerService().registerProcessingTimeTimer(fireTime);

                Integer currentCount = value.f1;
                Integer historyCount = counter.value();
                if (historyCount == null) {
                    historyCount = 0;
                }
                int totalCount = historyCount + currentCount;
                // 更新状态
                counter.update(totalCount);

            }

            // 当闹钟到了指定时间,就执行onTimer方法
            @Override
            public void onTimer(long timestamp, OnTimerContext ctx, Collector<Tuple2<String, Integer>> out) throws Exception {
                // 定时触发,输出当前结果
                Integer value = counter.value();
                String currentKey = ctx.getCurrentKey();
                // 输出k,v
                out.collect(Tuple2.of(currentKey, value));
            }
        });
        
        process.print();

        env.execute("timer job");

    }
}
案列二:基于eventTime 的Timer

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.api.common.typeinfo.TypeHint;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.tuple.Tuple;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;

/**
 * 只有keyedStream在使用ProcessFunction时可以使用State和Timer定时器
 */
public class ProcessFunctionWithTimerDemo2 {

    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
        //1000,hello
        DataStreamSource<String> lines = env.socketTextStream("localhost", 8888);

        SingleOutputStreamOperator<String> linesWithWaterMark = lines.assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor<String>(Time.seconds(0)) {
            @Override
            public long extractTimestamp(String element) {
                return Long.parseLong(element.split(",")[0]);
            }
        });

        SingleOutputStreamOperator<Tuple2<String, Integer>> wordAndOne = linesWithWaterMark.map(new MapFunction<String, Tuple2<String, Integer>>() {
            @Override
            public Tuple2<String, Integer> map(String line) throws Exception {
                String word = line.split(",")[1];
                return Tuple2.of(word, 1);
            }
        });

        //调用keyBy进行分组
        KeyedStream<Tuple2<String, Integer>, Tuple> keyed = wordAndOne.keyBy(0);

        //没有划分窗口,直接调用底层的process方法
        keyed.process(new KeyedProcessFunction<Tuple, Tuple2<String, Integer>, Tuple2<String, Integer>>() {

            private transient ListState<Tuple2<String, Integer>> bufferState;

            @Override
            public void open(Configuration parameters) throws Exception {
                ListStateDescriptor<Tuple2<String, Integer>> listStateDescriptor = new ListStateDescriptor<Tuple2<String, Integer>>(
                        "list-state",
                        TypeInformation.of(new TypeHint<Tuple2<String, Integer>>(){})
                );

                bufferState = getRuntimeContext().getListState(listStateDescriptor);
            }

            @Override
            public void processElement(Tuple2<String, Integer> value, Context ctx, Collector<Tuple2<String, Integer>> out) throws Exception {

                //out.collect(value);

                bufferState.add(value);
                //获取当前的event time
                Long timestamp = ctx.timestamp();

                //10:14:13   ->   10:15:00
                //输入的时间 [10:14:00, 10:14:59) 注册的定时器都是 10:15:00
                System.out.println("current event time is : " + timestamp);

                //注册定时器,如果注册的是EventTime类型的定时器,当WaterMark大于等于注册定时器的时间,就会触发onTimer方法
                long timer = timestamp - timestamp % 60000 + 60000;
                System.out.println("next timer is: " + timer);
                ctx.timerService().registerEventTimeTimer(timer);
            }

            @Override
            public void onTimer(long timestamp, OnTimerContext ctx, Collector<Tuple2<String, Integer>> out) throws Exception {

                Iterable<Tuple2<String, Integer>> iterable = bufferState.get();

                for (Tuple2<String, Integer> tp : iterable) {
                    out.collect(tp);
                }

                //请求当前ListState中的数据
                bufferState.clear();
            }
        }).print();
        env.execute();
    }
}
案列三:基于eventTime 的Timer

// 创建bean类CountWithTimestamp,里面有三个字段
public class CountWithTimestamp {
    public String key;
    public long count;
    public long lastModified;
}

// 创建FlatMapFunction的实现类Splitter,作用是将字符串分割后生成多个Tuple2实例,f0是分隔后的单词,f1等于1:
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.util.Collector;
import org.apache.flink.util.StringUtils;

public class Splitter implements FlatMapFunction<String, Tuple2<String, Integer>> {
    @Override
    public void flatMap(String s, Collector<Tuple2<String, Integer>> collector) throws Exception {

        if(StringUtils.isNullOrWhitespaceOnly(s)) {
            System.out.println("invalid line");
            return;
        }

        for(String word : s.split(" ")) {
            collector.collect(new Tuple2<String, Integer>(word, 1));
        }
    }
}

// 最后是整个逻辑功能的主体:ProcessTime.java,这里面有自定义的KeyedProcessFunction子类,还有程序入口的main方法import com.bolingcavalry.Splitter;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.java.tuple.Tuple;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.AssignerWithPeriodicWatermarks;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.watermark.Watermark;
import org.apache.flink.util.Collector;
import java.text.SimpleDateFormat;
import java.util.Date;


/**
 * @author will
 * @email zq2599@gmail.com
 * @date 2020-05-17 13:43
 * @description 体验KeyedProcessFunction类(时间类型是处理时间)
 */
public class ProcessTime {

    /**
     * KeyedProcessFunction的子类,作用是将每个单词最新出现时间记录到backend,并创建定时器,
     * 定时器触发的时候,检查这个单词距离上次出现是否已经达到10秒,如果是,就发射给下游算子
     */
    static class CountWithTimeoutFunction extends KeyedProcessFunction<Tuple, Tuple2<String, Integer>, Tuple2<String, Long>> {

        // 自定义状态
        private ValueState<CountWithTimestamp> state;

        @Override
        public void open(Configuration parameters) throws Exception {
            // 初始化状态,name是myState
            state = getRuntimeContext().getState(new ValueStateDescriptor<>("myState", CountWithTimestamp.class));
        }

        @Override
        public void processElement(
                Tuple2<String, Integer> value,
                Context ctx,
                Collector<Tuple2<String, Long>> out) throws Exception {

            // 取得当前是哪个单词
            Tuple currentKey = ctx.getCurrentKey();

            // 从backend取得当前单词的myState状态
            CountWithTimestamp current = state.value();

            // 如果myState还从未没有赋值过,就在此初始化
            if (current == null) {
                current = new CountWithTimestamp();
                current.key = value.f0;
            }

            // 单词数量加一
            current.count++;

            // 取当前元素的时间戳,作为该单词最后一次出现的时间
            current.lastModified = ctx.timestamp();

            // 重新保存到backend,包括该单词出现的次数,以及最后一次出现的时间
            state.update(current);

            // 为当前单词创建定时器,十秒后后触发
            long timer = current.lastModified + 10000;

            ctx.timerService().registerProcessingTimeTimer(timer);

            // 打印所有信息,用于核对数据正确性
            System.out.println(String.format("process, %s, %d, lastModified : %d (%s), timer : %d (%s)\n\n",
                    currentKey.getField(0),
                    current.count,
                    current.lastModified,
                    time(current.lastModified),
                    timer,
                    time(timer)));

        }

        /**
         * 定时器触发后执行的方法
         * @param timestamp 这个时间戳代表的是该定时器的触发时间
         * @param ctx
         * @param out
         * @throws Exception
         */
        @Override
        public void onTimer(
                long timestamp,
                OnTimerContext ctx,
                Collector<Tuple2<String, Long>> out) throws Exception {

            // 取得当前单词
            Tuple currentKey = ctx.getCurrentKey();

            // 取得该单词的myState状态
            CountWithTimestamp result = state.value();

            // 当前元素是否已经连续10秒未出现的标志
            boolean isTimeout = false;

            // timestamp是定时器触发时间,如果等于最后一次更新时间+10秒,就表示这十秒内已经收到过该单词了,
            // 这种连续十秒没有出现的元素,被发送到下游算子
            if (timestamp == result.lastModified + 10000) {
                // 发送
                out.collect(new Tuple2<String, Long>(result.key, result.count));

                isTimeout = true;
            }

            // 打印数据,用于核对是否符合预期
            System.out.println(String.format("ontimer, %s, %d, lastModified : %d (%s), stamp : %d (%s), isTimeout : %s\n\n",
                    currentKey.getField(0),
                    result.count,
                    result.lastModified,
                    time(result.lastModified),
                    timestamp,
                    time(timestamp),
                    String.valueOf(isTimeout)));
        }
    }


    public static void main(String[] args) throws Exception {
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // 并行度1
        env.setParallelism(1);

       // 处理时间
        env.setStreamTimeCharacteristic(TimeCharacteristic.ProcessingTime);

        // 监听本地9999端口,读取字符串
        DataStream<String> socketDataStream = env.socketTextStream("localhost", 9999);

        // 所有输入的单词,如果超过10秒没有再次出现,都可以通过CountWithTimeoutFunction得到
        DataStream<Tuple2<String, Long>> timeOutWord = socketDataStream
                // 对收到的字符串用空格做分割,得到多个单词
                .flatMap(new Splitter())
                // 设置时间戳分配器,用当前时间作为时间戳
                .assignTimestampsAndWatermarks(new AssignerWithPeriodicWatermarks<Tuple2<String, Integer>>() {

                    @Override
                    public long extractTimestamp(Tuple2<String, Integer> element, long previousElementTimestamp) {
                        // 使用当前系统时间作为时间戳
                        return System.currentTimeMillis();
                    }

                    @Override
                    public Watermark getCurrentWatermark() {
                        // 本例不需要watermark,返回null
                        return null;
                    }
                })
                // 将单词作为key分区
                .keyBy(0)
                // 按单词分区后的数据,交给自定义KeyedProcessFunction处理
                .process(new CountWithTimeoutFunction());

        // 所有输入的单词,如果超过10秒没有再次出现,就在此打印出来
        timeOutWord.print();

        env.execute("ProcessFunction demo : KeyedProcessFunction");
    }

    public static String time(long timeStamp) {
        return new SimpleDateFormat("yyyy-MM-dd hh:mm:ss").format(new Date(timeStamp));
    }
}

注意事项

优先级队列默认使用的是内存存储,在一些数据量比较大并且重度依赖定时触发的任务会占用比较大的内存,可以选择Rocksdb存储定时信息

flink为了保证定时触发操作(onTimer)与正常处理(processElement)操作的线程安全,做了同步处理,在调用触发时必须要获取到锁,也就是二者同时只能有一个执行,因此一定要保证onTimer处理的速度,以免任务发生阻塞。文章来源地址https://www.toymoban.com/news/detail-729254.html

如果不做同步处理,processElement方法中会进行state.update(),onTimer中会进行state.value(),两者会发生不一致从而引发线程安全问题。

到了这里,关于Flink timer定时器的文章就介绍完了。如果您还想了解更多内容,请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章,希望大家以后多多支持TOY模板网!

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处: 如若内容造成侵权/违法违规/事实不符,请点击违法举报进行投诉反馈,一经查实,立即删除!

领支付宝红包 赞助服务器费用

相关文章

  • Python中的定时器用法:Timer定时器和schedule库

    目录 一、引言 二、Timer定时器 1、Timer定时器的原理 2、Timer定时器的使用方法 3、Timer定时器的实际应用案例 三、schedule库 1、schedule库的原理 2、schedule库的使用方法 3、schedule库的实际应用案例 四、Timer定时器和schedule库的比较 1、功能差异 2、适用场景 五、实际应用案例 六、

    2024年01月16日
    浏览(71)
  • Flink的KeyedProcessFunction基于Event Time和Process Time的定时器用法实例分析

    在Flink底层,我们可以不定义任何具体的算子(比如 map,filter,或者 window),而只是提炼出一个统一的【处理】(process)操作——它是所有转换算子的一个概括性的表达,可以自定义处理逻辑,所以这一层接口就被叫作【处理函数】(process function)。在处理函数中,我们直

    2024年01月19日
    浏览(40)
  • Jmeter之同步定时器(Synchronizing Timer)

    同步定时器类似LoadRunner的集合点,作用是阻塞线程,达到指定的线程数量后,再一起释放。 添加定时器同步定时器(Synchronizing Timer) 1、模拟用户组的数量:每次释放的线程数量,即 并发数。 默认为0 设置为0则并发数等于线程租中的线程数;设置大于0则等待达到这个数量

    2024年02月11日
    浏览(55)
  • GD32系列笔记六:定时器Timer

    目录 一、定时器的作用 二、定时器介绍 三、定时器配置 1. 用作封装延时函数,提高程序实时性; 2. 测试某段代码的执行时间; 3. 一些外设的核心,如PWM输入捕获、输出比较等。 1.时钟树 2.结构图(基本定时器为例)  TIMER_CK就是CK_TIMER    3. 工作原理               1. 

    2024年02月13日
    浏览(42)
  • MSP432---Timer_A定时器详解

    msp432数据手册slau356h.pdf,P782 目录 1. 时钟源选择和分频器 2. 定时器A工作(计数)模式  2.1 停止模式  2.2 增计数模式  2.3 连续计数模式  2.4 增减计数模式 3. 捕获/比较模块  3.1 捕获模式 3.2 比较模式  4. 中断 5. Timer_A寄存器 6. 软件实例 MSP432单片机定时器模块 有:看门狗定时

    2024年02月16日
    浏览(49)
  • TM4C123系列(五)————timer定时器(timer模式)

    一.实验简介 通过定时器的timer模式来计时实现以1s为间隔将LED翻转。 二.板载定时器介绍 TM4C有两种定时器,一种为16/32bit的,一种是32/64bit的,两种定时器各有六个,对于每个定时器来说,它可以单独以较大的bit位作为一个定时器工作,也可以拆分为两个较小的bit位的定时器

    2024年02月15日
    浏览(39)
  • 物联网操作系统-软件定时器(software timer)

    软件定时器就是\\\"闹钟\\\",你可以设置闹钟, ⚫ 在 30 分钟后让你起床工作 ⚫ 每隔 1 小时让你例行检查机器运行情况 软件定时器也可以完成两类事情: ⚫ 在\\\"未来\\\"某个时间点,运行函数 ⚫ 周期性地运行函数 日常生活中我们可以定无数个\\\"闹钟\\\",这无数的\\\"闹钟\\\"要基于一个真实

    2024年02月21日
    浏览(41)
  • C#里面的三种定时计时器:Timer

    在.NET中有三种计时器: 1、System.Windows.Forms命名空间下的Timer控件,它直接继承自Componet。Timer控件只有绑定了Tick事件和设置Enabled=True后才会自动计时,停止计时可以用Stop()方法控制,通过Stop()停止之后,如果想重新计时,可以用Start()方法来启动计时器。Timer控件和它所在的

    2024年02月07日
    浏览(39)
  • RT-Thread 15. list_timer与软定时器

    1. 代码 2. list_timer结果 3. 解释 0x0000012c: 300个时钟周期,即处于rt_thread_mdelay(3000); 0x00000032: 50个时钟周期,即处于rt_thread_mdelay(500); 4.增加软定时器 //usrtimer.c 5. MSH命令开启软件定时器 timer_static_sample()会执行,timer1和timer2会开始工作。输入list_timer,发现多了timer1、timer2

    2024年01月19日
    浏览(40)
  • 开发一个RISC-V上的操作系统(七)—— 硬件定时器(Hardware Timer)

    目录 往期文章传送门 一、硬件定时器 硬件实现 软件实现 二、上板测试 开发一个RISC-V上的操作系统(一)—— 环境搭建_riscv开发环境_Patarw_Li的博客-CSDN博客 开发一个RISC-V上的操作系统(二)—— 系统引导程序(Bootloader)_Patarw_Li的博客-CSDN博客 开发一个RISC-V上的操作系统

    2024年02月12日
    浏览(41)

觉得文章有用就打赏一下文章作者

支付宝扫一扫打赏

博客赞助

微信扫一扫打赏

请作者喝杯咖啡吧~博客赞助

支付宝扫一扫领取红包,优惠每天领

二维码1

领取红包

二维码2

领红包