ProcessFunction和CoProcessFunction
 
说明
 
DataStream与KeyedStreamd都有Process方法,
 DataStream接收的是ProcessFunction，而KeyedStream接收的是KeyedProcessFunction(原本也支持ProcessFunction，现在已被废弃)
 
0.AbstractRichFunction介绍
 
1.ProcessFunction对flink更精细的操作
 
<1> Events（流中的事件）
 <2> State(容错，一致性，仅仅用于keyed stream)
 <3> Timers(事件时间和处理时间，仅仅适用于keyed stream)
 
ProcessFunction可以视为是FlatMapFunction，但是它可以获取keyed state和timers。每次有事件流入processFunction算子就会触发处理。
 
为了容错，ProcessFunction可以使用RuntimeContext访问flink内部的keyed state。
 
timer允许应用程序对处理时间和事件时间的变化做出反应。每次有事件到达都会调用函数processElement(...),该函数有参数，也就是Context对象，该对象可以访问元素的事件时间戳和TimerService，还有侧输出。
 
TimerService可用于注册为后续处理事件或者事件时间的回调。当达到计时器的特定时间时，将调用onTimer(...)方法。在该调用期间，所有状态再次限定为创建计时器的key，允许计时器操纵keyed状态。
 
2.CoProcessFunction 实现底层join
 
<1> 实现底层join操作典型模板就是：
 
为一个或者两个输入创建一个状态对象
根据输入的事件更新状态
根据从另一个流接受的元素，更新状态并且产生joined结果
3.KeyedProcessFunction
 
keyedProcessFunction是ProcessFunction的扩展，可以在onTimer获取timer的key (通过context.getCurrentKey方法)
 
4.Timer类型
 
1.两种类型(事件时间和处理时间)的timer都是由TimerService维护并且以队列的形式执行。
 
TimerService会使用key和timestamp对timer进行去重，也即是对于每一对key和timestamp仅仅会存在一个timer。如果同一个timestamp注册了多个timers，onTimer()函数仅仅会调用一次。
 
对于onTimer()和processElement()方法flink是做了同步的，所以不需要关系并发问题。
 
 
 
image.png
 
 
 
 
image.png
 
5.ProcessFunction与状态的结合使用案例
 
WordCount,如果某一个key一分钟(事件时间)没有更新,就直接输出。
 基本思路：
 // 1.ValueState内部包含了计数、key和最后修改时间
 // 2.对于每一个输入的记录,ProcessFunction都会增加计数,并且修改时间戳
 // 3.该函数会在事件时间的后续1min调度回调函数
 // 4.然后根据每次回调函数,就去检查回调事件时间戳和保存的时间戳,如果匹配就将数据发出
 
 
 
public class ProcessFunctionExample {

    // 1.ValueState内部包含了计数、key和最后修改时间
    // 2.对于每一个输入的记录,ProcessFunction都会增加计数,并且修改时间戳
    // 3.该函数会在事件时间的后续1min调度回调函数
    // 4.然后根据每次回调函数,就去检查回调事件时间戳和保存的时间戳,如果匹配就将数据发出


    private static class StreamDataSource extends RichParallelSourceFunction<Tuple3<String, Long, Long>> {

        private volatile boolean running = true;


        @Override
        public void run(SourceContext<Tuple3<String, Long, Long>> sourceContext) throws Exception {

            Tuple3[] elements = new Tuple3[]{
                    Tuple3.of("a", 1L, 1000000050000L),
                    Tuple3.of("a", 1L, 1000000054000L),
                    Tuple3.of("a", 1L, 1000000079900L),
                    Tuple3.of("a", 1L, 1000000115000L),
                    Tuple3.of("b", 1L, 1000000100000L),
                    Tuple3.of("b", 1L, 1000000108000L)
            };

            int count = 0;
            while (running && count < elements.length) {
                sourceContext.collect(new Tuple3<>((String) elements[count].f0, (Long) elements[count].f1, (Long) elements[count].f2));
                count++;
                Thread.sleep(10000);
            }

        }

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


    /**
     * 存储在状态中的对象
     */
    public static class CountWithTimestamp {
        //单词
        public String key;
        //单词计数
        public long count;
        //最近更新时间
        public long lastModified;

        @Override
        public String toString() {
            return "CountWithTimestamp{" +
                    "key='" + key + '\'' +
                    ", count=" + count +
                    ", lastModified=" + new Date(lastModified) +
                    '}';
        }
    }


    /**
     * ProcessFunction有两个泛型类,一个输入一个输出
     */
    public static class CountWithTimeoutFunction extends ProcessFunction<Tuple2<String, Long>, Tuple2<String, Long>> {

        private ValueState<CountWithTimestamp> state;

        //最先调用
        @Override
        public void open(Configuration parameters) throws Exception {
            //根据上下文获取状态
            state = getRuntimeContext().getState(new ValueStateDescriptor<CountWithTimestamp>("myState", CountWithTimestamp.class));
        }

        @Override
        public void processElement(Tuple2<String, Long> input, Context context, Collector<Tuple2<String, Long>> output) throws Exception {

            CountWithTimestamp current = state.value();
            if (current == null) {
                current = new CountWithTimestamp();
                current.key = input.f0;
            }

            //更新ValueState
            current.count++;
            //这里面的context可以获取时间戳
            //todo 此时这里的时间戳可能为null,如果设置的时间为ProcessingTime
            current.lastModified = context.timestamp();
            System.out.println("元素"+input.f0+"进入事件时间为：" + new Date(current.lastModified));
            state.update(current);

            //注册ProcessTimer,更新一次就会有一个ProcessTimer
            context.timerService().registerEventTimeTimer(current.lastModified + 9000);
            System.out.println("定时触发时间为："+new Date(current.lastModified + 9000));
        }

        //EventTimer被触发后产生的行为
        //todo 这里的timestamp是触发时间
        @Override
        public void onTimer(long timestamp, OnTimerContext ctx, Collector<Tuple2<String, Long>> out) throws Exception {


            //获取上次时间,与参数中的timestamp相比,如果相差等于60s 就会输出
            CountWithTimestamp res = state.value();
            System.out.println("当前时间为："+new Date(timestamp)+res);
            if (timestamp >= res.lastModified + 9000) {
                System.out.println("定时器被触发："+"当前时间为"+new Date(timestamp)+" 最近修改时间为"+new Date(res.lastModified));
                out.collect(new Tuple2<String, Long>(res.key, res.count));
            }


        }
    }


    //执行主类
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        DataStream<Tuple2<String, Long>> data = env.addSource(new StreamDataSource()).setParallelism(1)
                .assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor<Tuple3<String, Long, Long>>(Time.milliseconds(0)) {
                    @Override
                    public long extractTimestamp(Tuple3<String, Long, Long> input) {
                        return input.f2;
                    }
                }).map(new MapFunction<Tuple3<String, Long, Long>, Tuple2<String, Long>>() {
                    @Override
                    public Tuple2<String, Long> map(Tuple3<String, Long, Long> input) throws Exception {
                        return new Tuple2<>(input.f0, input.f1);
                    }
                });

        data.keyBy(0).process(new CountWithTimeoutFunction()).print();

        env.execute();

    }


}
 
这一步的结果是：
 
 
 
 
image.png
 
发现共有四个OnTimer被执行，其中没有执行OnTimer的两条元素是
 
 
 
 
image.png
 
这两条消息定时器预计执行时间都超过了09:48:35,因为这个案例采用的是事件时间，而这六条元素最大的事件时间为09:48:35，所以默认到09:48:35就停止了
 
注意：看代码可以发现这里发送的元素之间是每隔10秒发送，因为以为会影响结果，实际是我们使用的是EventTime,所以OnTimer被执行的时间，是看事件时间。
 
如果将最大事件时间改一下，改成
 
 
 
 
image.png
 
结果就是除了他自身，其余onTimer全部被执行了，因为它的事件时间，超过了其余5个元素的定时器触发时间。
 
并且我们发现有一条消息满足了其中的条件。
 
这里有一个疑问就是：为什么a的所有最近修改时间都是09:48:45 ，a的最大事件时间？？？？
 分析可能是构造的数据源的原因。这里模拟的是将优先数据源作为无限数据源使用
 
解决问题：
 
 
 
 
 
 
一开始没有设置为EventTime,所以在处理的时候还是以Process Time来处理的。
 改完之后的效果：
 
 
分析问题产生的原因：因为一开始未指定时间类型为EventTime，所以默认是以Process Time来处理，而一般来说使用ProcessTime，就不需要指定Watermark了（Watermark只是与EventTime配合使用），但是代码中偏偏还是使用了assign...方法，所以会在数据加载完了，使用最近的元素的时间，生成一个Watermark，这时候有了Watermark才会执行onTimer方法，所以才会出现数据全部加载完，才执行onTimer方法；
 
而当指定为EventTime时，来一个元素就会生成一个Watermark，当Watermark大于某个元素的触发时间，OnTimer就会执行，而不是等数据全部加载完之后才会生成
 
所以上面一开始对某些onTimer没有执行的理解是错误的，应该按照上面没有指定EventTime的方式去理解
 

 最终代码为：
 
package com.meituan.flink.example;

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.api.java.tuple.Tuple3;
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.ProcessFunction;
import org.apache.flink.streaming.api.functions.source.RichParallelSourceFunction;
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;

import java.util.Date;

/**
 * @author ranbo
 * @version V1.0
 * @Title:
 * @Package local.example
 * @Description:
 * @date 2020-04-06 19:39
 */
public class ProcessFunctionExample {

    // 1.ValueState内部包含了计数、key和最后修改时间
    // 2.对于每一个输入的记录,ProcessFunction都会增加计数,并且修改时间戳
    // 3.该函数会在事件时间的后续1min调度回调函数
    // 4.然后根据每次回调函数,就去检查回调事件时间戳和保存的时间戳,如果匹配就将数据发出


    private static class StreamDataSource extends
        RichParallelSourceFunction<Tuple3<String, Long, Long>> {

        private volatile boolean running = true;


        @Override
        public void run(SourceContext<Tuple3<String, Long, Long>> sourceContext) throws Exception {

            Tuple3[] elements = new Tuple3[]{
                Tuple3.of("a", 1L, 1000000050000L), // 09:47:30
                Tuple3.of("a", 1L, 1000000054000L), // 09:47:34
                Tuple3.of("a", 1L, 1000000079900L), // 09:47:59
//                Tuple3.of("a", 1L, 1000000115000L), // 09:48:35

                Tuple3.of("b", 1L, 1000000100000L), // 09:48:20
                Tuple3.of("b", 1L, 1000000108000L),  // 09:48:28
                Tuple3.of("a", 1L, 1000000125000L) // 09:48:45
            };

            int count = 0;
            while (running && count < elements.length) {
                sourceContext.collect(new Tuple3<>((String) elements[count].f0, (Long) elements[count].f1, (Long) elements[count].f2));
                count++;
                Thread.sleep(3000);
            }

        }

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


    /**
     * 存储在状态中的对象
     */
    public static class CountWithTimestamp {
        //单词
        public String key;
        //单词计数
        public long count;
        //最近更新时间
        public long lastModified;

        @Override
        public String toString() {
            return "CountWithTimestamp{" +
                "key='" + key + '\'' +
                ", count=" + count +
                ", lastModified=" + new Date(lastModified) +
                '}';
        }
    }


    /**
     * ProcessFunction有两个泛型类,一个输入一个输出
     */
    public static class CountWithTimeoutFunction extends
        ProcessFunction<Tuple2<String, Long>, Tuple2<String, Long>> {

        private ValueState<CountWithTimestamp> state;

        private static int cnt = 0;
        //最先调用
        @Override
        public void open(Configuration parameters) throws Exception {
            //根据上下文获取状态
            state = getRuntimeContext().getState(new ValueStateDescriptor<CountWithTimestamp>("myState", CountWithTimestamp.class));
        }

        @Override
        public void processElement(Tuple2<String, Long> input, Context context, Collector<Tuple2<String, Long>> output) throws Exception {

            CountWithTimestamp current = state.value();
            if (current == null) {
                current = new CountWithTimestamp();
                current.key = input.f0;
            }

            //更新ValueState
            current.count++;
            //这里面的context可以获取时间戳
            //todo 此时这里的时间戳可能为null,如果设置的时间为ProcessingTime
            current.lastModified = context.timestamp();
            System.out.println(cnt + " 元素"+input.f0+"进入事件时间为：" + new Date(current.lastModified));
            cnt++;
            state.update(current);

            //注册ProcessTimer,更新一次就会有一个ProcessTimer
            context.timerService().registerEventTimeTimer(current.lastModified + 9000);
            System.out.println("定时触发时间为："+new Date(current.lastModified + 9000));
        }

        //EventTimer被触发后产生的行为
        //todo 这里的timestamp是触发时间
        @Override
        public void onTimer(long timestamp, OnTimerContext ctx, Collector<Tuple2<String, Long>> out) throws Exception {


            //获取上次时间,与参数中的timestamp相比,如果相差等于60s 就会输出
            CountWithTimestamp res = state.value();
            System.out.println("当前时间为："+new Date(timestamp)+res);
            if (timestamp == res.lastModified + 9000) {
//            if (timestamp >= res.lastModified + 9000) {
                System.out.println("定时器被触发："+"当前时间为"+new Date(timestamp)+" 最近修改时间为"+new Date(res.lastModified));
                out.collect(new Tuple2<String, Long>(res.key, res.count));
            } else {
                System.out.println("timestamp:" + timestamp + " res.lastModified:" + res.lastModified +" 当前时间为"+new Date(timestamp)+" 最近修改时间为"+new Date(res.lastModified));
            }
        }
    }


    //执行主类
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
        DataStream<Tuple2<String, Long>> data = env.addSource(new StreamDataSource()).setParallelism(1)
            .assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor<Tuple3<String, Long, Long>>(
                Time.milliseconds(0)) {
                @Override
                public long extractTimestamp(Tuple3<String, Long, Long> input) {
                    return input.f2;
                }
            }).map(new MapFunction<Tuple3<String, Long, Long>, Tuple2<String, Long>>() {
                @Override
                public Tuple2<String, Long> map(Tuple3<String, Long, Long> input) throws Exception {
                    return new Tuple2<>(input.f0, input.f1);
                }
            });

        data.keyBy(0).process(new CountWithTimeoutFunction()).print();

        env.execute();

    }


}
 

  

目录
 
 
 
场景：
 
onTimer 使用关键流程
 
数据处理流程
 
代码交互流程
 
onTimer 延迟数据处理的优劣
 
优点：
 
缺点：
 
onTimer 编码实践
 
业务场景描述
 
代码
 

场景：
 
某些特殊业务场景需要延迟数据处理，比如乱序数据。
 
某些业务场景只需要保留最新数据，中间更新过程忽略不计，比如客服问卷最新状态。
 
某些业务场景需要结合最近一段时间的数据进行处理，比如客服侧由于系统短时间单条数据更新多个字段，系统侧更新字段顺序错误，导致中间记录的 binlog 数据可能有误，需要结合一段时间的数据进行修正数据。
 
 
 
 
 
 
 
onTimer 使用关键流程
 
数据处理流程
 
 
key by 分区，把相同 key 的数据发送到同一个分区。
 
单条数据会设置定时器，可以让单个分区里面多条数据的定时器时间一样，那就只触发一次定时器。
 
 
单条数据可以更新 state，然后定时器触发时，可以再取 state 的数据进行业务逻辑处理。
 
 
 
代码交互流程
 
 
 
 
 
processElement 单条数据处理
 
 
onTimer 定时器回调
 
 
Managed State 缓存状态
 
 
TimerService 时间服务
 
 
 
 
 
onTimer 延迟数据处理的优劣
 
优点：
 
 
利用自带的 state，丰富的数据结构 ValueState 存最新数据、ListState 最一段时间数据、MapState 存 kv 数据。
 
 
下发数据灵活：定时器下发数据，每条数据都可以注册一个定时器下发一次数据，多条数据的定时器也可以合并成一个定时器下发一次数据。
 
 
定时器触发时数据处理非常灵活：可以取到 state 里面的所有数据，然后根据业务的逻辑进行数据处理。
 
 
下发数据量减少：比如2s延迟，多条数据合并成一个定时器，如果业务只要最新状态，就只需要下发一条数据。
 
 
 
 
 
缺点：
 
 
数据会有一定延迟。
 
 
 
 
 
 
 
onTimer 编码实践
 
业务场景描述
 
客服问卷只要最新状态的情况，比如问卷评价的数据一定在问卷未评价的数据之后，有时候问卷已评价和未评价的数据时间相近，下游消费时会乱序，进而发生未评价的数据更新已评价的数据，数据库里面问卷最新状态就是未评价了。
 
通过 onTimer 延迟5s数据处理，把5s内的数据统一处理，然后只保留最新问卷下发，比如5s内有未评价和已评价两条数据，那就只下发一条已评价数据就行，下游消费就不会乱序了。
 
 
 
代码
 
1.按照问卷 id 进行 keyBy 分区。
 
        DataStream<String> npsMainStream = npsMainEntityDataStream
            .keyBy(new KeySelector<NpsMainEntity, String>() {
                @Override public String getKey(NpsMainEntity npsMainEntity) throws Exception {
                    String key = npsMainEntity.getId() + "";
                    return key;
                }
            }).process(new StateTimerForOutOrderDataFunction());
 
2.processElement 方法里面每条数据来更新 state，这里有两种情况：
 
 
ValueState 为空会被更新。
 
 
已评价数据会更新 ValueState。
 
processElement 里面每条数据会注册一个定时器，ctx.timerService().registerProcessingTimeTimer(triggerTime);
 
3.onTimer 方法里面定时器触发，可以查询 state 进行业务逻辑处理
 
​
 
    /** 延迟下发数据，解决并发乱序问题 */
    static class StateTimerForOutOrderDataFunction extends KeyedProcessFunction<String, NpsMainEntity, String> {

        private final static int ttlMinute = 1000 * 5;
        private ValueState<NpsMainEntity> sendCacheStorage;

        @Override
        public void open(Configuration parameters) throws Exception {
            sendCacheStorage = getRuntimeContext().getState(new ValueStateDescriptor<NpsMainEntity>("CacheStorage", NpsMainEntity.class));
        }


        @Override public void processElement(NpsMainEntity npsMainEntity, Context ctx, Collector<String> out)
            throws Exception {
            try {
                // 每条 record 注册事件 ttlMinute 代表延迟时间
                /** 状态为空，或者状态是未评价，5s时间解决乱序问题 */
                if (sendCacheStorage.value() == null || sendCacheStorage.value().getEvaluationStatus() == 2
                    || npsMainEntity.getEvaluationCnt() == 1) {
                    sendCacheStorage.update(npsMainEntity);
                    writeSampleLog("update:" + npsMainEntity.toString(), NumberEnum.SEVEN);
                    // 合并定时器，秒级触发
                    long triggerTime = (System.currentTimeMillis() + ttlMinute) /1000 * 1000;
                    ctx.timerService().registerProcessingTimeTimer(triggerTime);
                }
                // 相关操作
            } catch (Exception e) {
                logger.error(e.getMessage());
            }
        }
      
        // 到达时间点触发事件
        @Override public void onTimer(long timestamp, OnTimerContext ctx, Collector<String> out)
            throws Exception {
            String onTimerValue = JSON.toJSONString(sendCacheStorage.value());
            out.collect(onTimerValue);
            writeSampleLog("onTimer onTimerValue:" + onTimerValue, NumberEnum.ZERO);
        }

    }
 
 

在MFC中我们经常需要定时读写数据，或者定时刷新界面，更新数据和状态，这就需要用到定时器，其实现函数为OnTimer，下面对其用法步骤（基于VS2010）进行简要说明：
 1、在类视图中点击需要使用定时器的对话框类，在属性窗口选择消息，在下面的消息函数中选中WM_TIMER，后面选择添加OnTimer函数，如下图所示： 
 
二、
 
本节继续讲与时间有关的定时器。定时器并不是一个类，主要考虑到，提起时间的话就不能不说定时器，所以就把它放到CTime和CTimeSpan之后讲解。
 
       定时器简介
 
       定时器，可以帮助开发者或者用户定时完成某项任务。在使用定时器时，我们可以给系统传入一个时间间隔数据，然后系统就会在每个此时间间隔后触发定时处理程序，实现周期性的自动操作。例如，我们可以在数据采集系统中，为定时器设置定时采集时间间隔为1个小时，那么每隔1个小时系统就会采集一次数据，这样就可以在无人操作的情况下准确的进行操作。
 
       MFC定时器
 
       VS2010编程中，我们可以使用MFC的CWnd类提供的成员函数SetTimer实现定时器功能，也可以使用Windows API函数SetTimer来实现。两者使用方法实际上很类似，但也有不同。
 
       CWnd类的SetTimer成员函数只能在CWnd类或其派生类中调用，而API函数SetTimer则没有这个限制，这是一个很重要的区别。因为本教程主要是讲解MFC编程，所以这里就先重点讲解MFC定时器的用法，关于API函数SetTimer的用法鸡啄米会在MFC定时器讲解的基础上进行延伸。
 
       鸡啄米下面分步骤给出使用MFC定时器的方法。
 
       1、启动定时器。
 
       启动定时器就需要使用CWnd类的成员函数SetTimer。CWnd::SetTimer的原型如下：
 
       UINT_PTR SetTimer(
              UINT_PTR nIDEvent,
              UINT nElapse,
              void (CALLBACK* lpfnTimer
        )(HWND,
           UINT,
           UINT_PTR,
           DWORD
        ) 
        );
 
       参数nIDEvent指定一个非零的定时器ID；参数nElapse指定间隔时间，单位为毫秒；参数lpfnTimer指定一个回调函数的地址，如果该参数为NULL，则WM_TIMER消息被发送到应用程序的消息队列，并被CWnd对象处理。如果此函数成功则返回一个新的定时器的ID，我们可以使用此ID通过KillTimer成员函数来销毁该定时器，如果函数失败则返回0。
 
       通过SetTimer成员函数我们可以看出，处理定时事件可以有两种方式，一种是通过WM_TIMER消息的消息响应函数，一种是通过回调函数。
 
       如果要启动多个定时器就多次调用SetTimer成员函数。另外，在不同的CWnd中可以有ID相同的定时器，并不冲突。
 
       2、为WM_TIMER消息添加消息处理函数，或者定义回调函数。
 
       如果调用CWnd::SetTimer函数时最后一个参数为NULL，则通过WM_TIMER的消息处理函数来处理定时事件。添加WM_TIMER消息的处理函数的方法是，在VS2010工程的Class View类视图中找到要添加定时器的类，点击右键，选择Properties，显示其属性页，然后在属性页工具栏上点击Messages按钮，下面列表就列出了所有消息，找到WM_TIMER消息，添加消息处理函数。添加后，cpp文件中会出现类似如下内容：
 
C++代码
 
BEGIN_MESSAGE_MAP(CExample44Dlg, CDialogEx)   
    ......   
    ON_WM_TIMER()   
END_MESSAGE_MAP()   
  
void CExample44Dlg::OnTimer(UINT_PTR nIDEvent)   
{   
    // TODO: Add your message handler code here and/or call default   
  
    CDialogEx::OnTimer(nIDEvent);   
}  
       之后就可以在OnTimer函数中进行相应的处理了。OnTimer的参数nIDEvent为定时器ID，即在SetTimer成员函数中指定的定时器ID，如果有多个定时器，我们可以像下面这样处理：
 
C++代码
 
void CExample44Dlg::OnTimer(UINT_PTR nIDEvent)      
{      
    // TODO: Add your message handler code here and/or call default      
    switch (nIDEvent)      
    {      
    case 1:      
        // 如果收到ID为1的定时器的消息则调用func1函数      
        func1();      
        break;      
    case 2:      
        // 如果收到ID为2的定时器的消息则调用func2函数      
        fun2();    
       break;     
    ......      
    default:      
        break;      
    }      
     
    CDialogEx::OnTimer(nIDEvent);      
}     
       如果调用CWnd::SetTimer函数时最后一个参数不为NULL，则需要定义回调函数。回调函数的形式如下：
 
C++代码
 
void CALLBACK EXPORT TimerProc(   
  
HWND hWnd, // handle of CWnd that called SetTimer   
  
UINT nMsg, // WM_TIMER   
  
UINT nIDEvent // timer identification   
  
DWORD dwTime // system time   
  
);  
       参数hWnd为调用SetTimer成员函数的CWnd对象的句柄，即拥有此定时器的窗口的句柄；参数nMsg为WM_TIMER，而且总是为WM_TIMER；参数nIDEvent为定时器ID；参数dwTime为系统启动以来的毫秒数，即GetTickCount函数的返回值。
 
       这样CWnd::SetTimer函数最后一个参数就可以为TimerProc。
 
       这里注意下，回调函数的名称不一定为TimerProc，可以取其他名字，但返回值类型、参数的类型和个数不能改变。
 
       鸡啄米给出一个回调函数的例子：
 
C++代码
 
void CALLBACK EXPORT TimerProc(HWND hWnd,UINT nMsg,UINT nTimerid,DWORD dwTime)      
{      
   switch(nTimerid)      
   {      
   case 1:       
         // 处理ID为1的定时器的事件      
         func1();      
         break;      
   case 2:       
         // 处理ID为2的定时器的事件      
         func2();      
         break;    
   ......   
   default:   
        break;     
   }      
}     
       回调函数为全局函数，需要写在使用它的位置的前面，或者写在后面然后在使用之前声明。
 
       3、销毁定时器。
 
       不再使用定时器时，可以销毁它。销毁定时器需使用CWnd类的KillTimer成员函数，CWnd::KillTimer函数的原型如下：
 
C++代码
 
BOOL KillTimer(UINT_PTR nIDEvent);  
       参数nIDEvent为要销毁的定时器的ID，是调用CWnd::SetTimer函数时设置的定时器ID。如果定时器被销毁则返回TRUE，而如果没有找到指定的定时器则返回FALSE。
 
       如果要销毁多个定时器，则多次调用KillTimer函数并分别传入要销毁的定时器的ID。
 
       通过Windows API函数使用定时器
 
       如果我们不使用MFC定时器，而通过Windows API函数使用定时器，其实是很类似的。下面鸡啄米简单说下步骤吧。
 
       1、启动定时器。
 
       使用API函数SetTimer启动定时器，SetTimer函数的原型如下：
 
C++代码
 
UINT_PTR SetTimer(         
    HWND    
            hWnd,   
    UINT_PTR    
            nIDEvent,   
    UINT    
            uElapse,   
    TIMERPROC    
            lpTimerFunc   
);  
       参数hWnd为与定时器关联的窗口的句柄；参数nIDEvent为非零的定时器ID，如果hWnd等于NULL，且还不存在ID为nIDEvent的定时器，那么nIDEvent参数被忽略，然后生成一个新ID的定时器，而如果hWnd不为NULL，且hWnd指定的窗口已存在ID为nIDEvent的定时器，那么这个已存在的定时器被新定时器所取代。参数uElapse和lpTimerFunc同CWnd::SetTimer函数。
 
       2、为WM_TIMER消息添加消息处理函数，或者定义回调函数。
 
       如果调用SetTimer函数时最后一个参数为NULL，我们需要自己为WM_TIMER消息添加处理函数，要注意的是，WM_TIMER消息的附加数据wParam为定时器ID，lParam为回调函数的指针，如果调用SetTimer时回调函数为NULL，那么lParam也为NULL。
 
       而如果调用SetTimer函数时最后一个参数不为NULL，我们就需要定义回调函数。回调函数的定义同MFC定时器。
 
       3、销毁定时器。
 
       销毁定时器使用KillTimer API函数，原型如下：
 
C++代码
 
BOOL KillTimer(HWND hWnd,UINT_PTR uIDEvent);   
       参数hWnd为与定时器关联的窗口的句柄，与启动定时器时SetTimer函数的hWnd参数值相同；参数uIDEvent为要销毁的定时器的ID，如果传递给SetTimer的参数hWnd有效，则uIDEvent应与传递给SetTimer的参数nIDEvent相同，而如果SetTimer的参数hWnd为NULL，则uIDEvent应为SetTimer返回的定时器ID。该函数成功则返回TRUE，否则返回FALSE。
 
       MFC定时器应用实例
 
       鸡啄米给大家演示一个定时器的例子，该实例功能很简单，就是使用两个定时器，定时更新两个编辑框中的显示内容，第一个编辑框每秒刷新一次，从1刷新到10，然后销毁定时器，第二个编辑框每两秒刷新一次，从1刷新到5，然后销毁定时器。下面简单说下步骤：
 
       1、创建基于对话框的工程，名称设为“Example44”。
 
       2、在自动生成的对话框模板IDD_EXAMPLE44_DIALOG中，删除“TODO: Place dialog controls here.”静态文本控件。添加两个静态文本框控件，Caption分别设为“1秒钟刷新一次”和“2秒钟刷新一次”，再添加两个个Edit Control控件，ID使用默认的IDC_EDIT1和IDC_EDIT2，两者的Read Only属性都设为True。此时的对话框模板如下图：
 
 
       3、为CExample44Dlg类添加两个成员变量，分别为m_nData1、m_nData2，并在CExample44Dlg类的构造函数中初始化：
 
C++代码
 
CExample44Dlg::CExample44Dlg(CWnd* pParent /*=NULL*/)   
    : CDialogEx(CExample44Dlg::IDD, pParent)   
{   
    m_hIcon = AfxGetApp()->LoadIcon(IDR_MAINFRAME);   
    // 两个数据初始化为0   
    m_nData1 = 0;   
    m_nData2 = 0;   
}  
       4、在对话框模板上双击OK按钮，添加点击消息的处理函数，并修改如下：
 
C++代码
 
void CExample44Dlg::OnBnClickedOk()   
{   
    // TODO: Add your control notification handler code here   
    // 启动ID为1的定时器，定时时间为1秒   
    SetTimer(1, 1000, NULL);   
    // 启动ID为2的定时器，定时时间为2秒   
    SetTimer(2, 2000, NULL);   
  
    //CDialogEx::OnOK();   
}  
       这样，点击OK按钮时就不会退出，而是启动两个定时器。
 
       5、根据上面MFC定时器讲解中为WM_TIMER消息添加处理函数的方法，添加WM_TIMER的消息处理函数OnTimer，并修改其实现如下：
 
C++代码
 
void CExample44Dlg::OnTimer(UINT_PTR nIDEvent)   
{   
    // TODO: Add your message handler code here and/or call default   
    switch (nIDEvent)   
    {   
    case 1:   
        // 如果m_nData1已经达到10，则销毁ID为1的定时器   
        if (10 == m_nData1)   
        {   
            KillTimer(1);   
            break;   
        }   
        // 刷新编辑框IDC_EDIT1的显示   
        SetDlgItemInt(IDC_EDIT1, ++m_nData1);   
        break;   
    case 2:   
        // 如果m_nData2已经达到5，则销毁ID为2的定时器   
        if (5 == m_nData2)   
        {   
            KillTimer(2);   
            break;   
        }   
        // 刷新编辑框IDC_EDIT2的显示   
        SetDlgItemInt(IDC_EDIT2, ++m_nData2);   
    default:   
        break;   
    }   
  
    CDialogEx::OnTimer(nIDEvent);   
}  
       6、运行程序，点击OK按钮，查看效果。
 
 
       关于定时器的内容就讲这些，相信了解了这些，一般的定时器应用都能解决了。

如果遇见一下问题，可能会有帮助：
 
OnTimer进不去
非静态成员引用必须与特定对象相对报错
不知道怎么开启和设置定时器
 
首先：用SetTimer开启定时器
 
	//设置定时器更新界面
	SetTimer(0, 1000, NULL);                       //1000为时间间隔
 
其次：在这里设置响应消息（不然会进不去OnTimer函数对应第一个问题）
 
BEGIN_MESSAGE_MAP(CFIRADlg, CDialogEx)
	
	ON_WM_TIMER()
	
END_MESSAGE_MAP()
 
最后，申明和书写OnTimer函数（注意这里面对应第二个问题）
 （CDialogEx::OnTimer(nIDEvent);这一句出现了两次，需要注意的是：这个CDialogEx是和MFC程序的类型有关，最开始我写的CFrameWnd会报错，因为参考了另一个程序，所以有问题，后面改成CDialogEx就好了，因为我的程序是基于对话框的，创建MFC程序的时候会让我们选择基于什么，CFrameWnd应该是基于单个或者多个文档的）
 
看一看创建时的选项： 
 
 
afx_msg void OnTimer(UINT_PTR nIDEvent);//打开之后定时器函数每隔一段时间执行一次
 
//----------------------------------------------------------------------------------
/**
\brief    Timer函数计算帧率
\param    nIDEvent  定时器ID

\return 无
*/
//----------------------------------------------------------------------------------
void CFIRADlg::OnTimer(UINT_PTR nIDEvent)
{
	// TODO: Add your message handler code here and/or call default
	
	if ((left.m_bIsSnap == 0) && (right.m_bIsSnap == 0))
	{
		CDialogEx::OnTimer(nIDEvent);
		return;
	}

	if (left.m_bIsSnap != 0)
	{
      left.m_pCamsFps.UpdateFps();
	  left.fFps = (float)left.m_pCamsFps.GetFps();
	}

	if (right.m_bIsSnap != 0)
	{
		right.m_pCamsFps.UpdateFps();
		right.fFps = (float)right.m_pCamsFps.GetFps();
	}

	CDialogEx::OnTimer(nIDEvent);
}