一只安静的猫

一、引言

在一个比较大的系统当中，很多地方都需要定时功能，比如某些存在内存中的数据，需要定时的做持久化（实时持久化可能带来很大的性能问题），如果这个系统要求跨平台。那么实现一个跨平台的定时器就很有必要了。
c++11的标准中，实现了多线程,头文件为<thread>,也有了系统时间相关的库，头文件为<chrono>,这使得使用c++11实现跨平台的定时器有了可能。在这之前，多线程一般都是使用操作系统相关的库。

二、 定时器的基本功能

在定时器的实现之前，需要事先计划好一个定时器的基本功能。大概的功能如下：
1.添加一个定时事件
2.删除一个指定的定时事件
3.定时事件支持只执行一次或者重复执行
4.定时器启动指令
这个部分的内容在Linux 下定时器的实现方式分析中有很详细的介绍

三、 数据结构知识预备

在实现定时器中，需要有一个容器去存储所有的定时事件，比如说最容易想到的数组，如果我们用数组保存定时事件，那么在找出当前需要执行的定时事件的伪代码就是

    for(i = 0 ; i < events_array_size; ++i){
        if(events[i].time <= current_time){
            exec_in_child_thread(events);
        }
    }

这里就是遍历所有的事件，将已经到了执行时间的事件放到子线程中去执行。

在这里有一个可以优化效率问题：注意到每一次查询是否有事件到了执行时间，都会遍历整个数组，也就是O(n)的复杂度，并且查询周期也是非常短的，所以会有非常大的性能问题。因此，我们可以将这里的数组变成有序的，每一次只需查询第一个事件，如果还没到执行时间，那么之后的事件就不用遍历了。这样平均下来就是O(1)的时间复杂度。考虑到我们在插入，删除时始终要保证数组的有序，可以采用最小堆。也就是堆顶始终是最先到执行时间的一项。

四、c++11多线程的知识预备

4.1 join()还是detach()

c++11中创建一个线程很容易

#include <iostream>
#include <thread>

void function_1(){
    std::cout << "hello world" << std::endl;
    while(true){}
}

int main(){
    std::thread t1(function_1);
    t1.join();      //加入运行

    std::cout << "hahah" << std::endl;

    return 0;
}

这段代码就是创建了一个t1的线程，来执行function_1函数。注意到我们这里使用了t1.join(),字面上的意思是将子线程加入的主线程来执行，这样导致的结果是主线程会在t1.join()这里停止，直到t1执行结束后再往下执行。在这里function_1中存在死循环，所以hahah永远不会被输出。而如果使用t1.detach()的话，即将子线程从主线程分离出去，那么主线程就会一直向下执行，在这里就会执行到return 0,然后主线程销毁，detach出去的子线程可能就没有任何显示的机会了（查过资料说detach出去子线程的行为的undefined的）。

在这个定时器的实现中，当然不能使用join(),因为会阻塞程序的正常运行。

4.2 如何使得定时器的使用是线程安全的。

在一个程序中，可能很多地方会使用到定时器去触发某个动作，但是为了效率，肯定不会创建多个定时器实例，因此定时器的实现应该是单例的。因为需要单例，因此定时器的创建，定时器中定时事件的添加和删除都是线程安全的（因为同一时间可能会有多个线程操作定时器）。

4.2.1 线程安全的单例模式

单例模式的实现方法有很多种，因为能力问题，不对这一点多做叙述。可以看这一篇单例模式(Singleton)及其C++实现。

我这里使用的是使用static来创建局部可见的全局变量

static Timer* getInstance(std::chrono::milliseconds tick){
    static Timer timer(tick);
    return &timer;
}

4.2.2 使用锁保证定时事件的添加和删除是线程安全的。

在往定时器中添加和删除定时事件时，就是在做最小堆的添加和删除。这个中间线程不安全的现象为:以添加为例，往最小堆数组末尾加入一个数据，然后再逐层向上调整。如果在这个调整的过程中，另外一个线程执行了删除操作，两个调整过程就可能同时发生，造成不可预知的错误，使得最小堆并不有序。

所以添加和删除操作应该是互斥的，只有其中一个操作执行结束，另一个操作才可以开始。c++中可以使用信号量mutex和锁locker轻松的完成互斥的需求。示例代码如下：

#include <iostream>
#include <thread>
#include <string>
#include <mutex>
#include <fstream>

class LofFile{
    public:
        LofFile(){
            f.open("log.txt");
        }

        void shared_print(std::string id,int value){
            std::lock(m_mutex,m_mutex2);
            std::lock_guard<std::mutex> locker(m_mutex,std::adopt_lock);
            std::lock_guard<std::mutex> locker2(m_mutex2,std::adopt_lock);
            std::cout << "from" << id << ":" << value << std::endl;
        }

        void shared_print2(std::string id,int value){
            std::lock(m_mutex,m_mutex2);
            std::lock_guard<std::mutex> locker2(m_mutex2,std::adopt_lock);      //lock_guard是为了防止语句执行中出现异常，锁不被释放。这样做可以保证在退出代码块后解锁
            std::lock_guard<std::mutex> locker(m_mutex,std::adopt_lock);
            std::cout << "from" << id << ":" << value << std::endl;
        }

    protected:
    private:
        std::mutex m_mutex;     //使用信号量来解决资源竞争
        std::mutex m_mutex2;
        std::ofstream f;

};

void function_1(LofFile& log){
    for(int i = 0; i > -100; i--){
        log.shared_print("From t1:",i);
    }
};

int main(){
    LofFile log;
    std::thread t1(function_1,std::ref(log));//线程开始运行
    for(int i = 0; i < 100; i++){
        log.shared_print2("From main",i);
    }

    t1.join();
    return 0;
}

五、总结与实现

分析到这里，一个定时器的基本要素已经具备了。
1.用来存储定时事件，并且添加和删除操作都是线程安全的最小堆
2.可以多线程实现定时事件的执行从而不阻塞主线程。

下面贴出所有的实现代码：

5.1 最小堆的实现。

SortedHeap.hpp

/**
 * 排序堆的实现
 * 堆使用的是完全二叉树的结构，使用数组（数组从零开始）保存，那么最后一个非叶子节点为(n - 1) / 2
 * 堆的构建一直都是在有序的基础上的，那么每次调整只需比较i和(i - 1) / 2的元素，依次上推
 * 支持任意类的排序
 * 当前还不支持多线程环境下的使用
 * author:jiangpengfie
 * date:2017-05-09
 */
#ifndef SORTEDHEAP_H
#define SORTEDHEAP_H
#include <iostream>
#include <vector>
#include <functional>
#include <memory>
#include <mutex>
#include <condition_variable>
#include "src/core/util/util.h"


template<class T>
class SortedHeap{
    private:
        struct HeapNode{
            unsigned int id;
            T obj;
            HeapNode(unsigned int id,T t):obj(t){
                this->id = id;
            }
        };
        std::vector<HeapNode> heap;
        unsigned int autoIncrementId;
        std::function<bool(T& ,T&)> cmp;    //比较函数，实现选择构造最大堆还是最小堆
        std::mutex mu1;                
        std::mutex mu2;                

        /**
         * 插入节点后调整堆中不符合的节点
         */
        void adjustAfterInsert();

        /**
         * pop出堆顶元素后调整堆中不符合的节点
         */
        void adjustAfterPopTop();

        /**
         * 删除节点后调整堆中不符合的节点
         * @param i 删除的节点id
         */
        void adjustAfterDelete(int id);         

        void swap(HeapNode& t1,HeapNode& t2);

        void deleteNodeByPos(const unsigned int pos);
    public:
        /**
         * 构造函数
         * @param cmp 用来比较
         */
        SortedHeap(std::function<bool(T&,T&)> cmp);
        /**
         * 插入节点
         * @param node 插入的节点
         */
        unsigned int insertNode(T& node);
        /**
         * 删除节点，时间复杂度为O(n)
         * @param id  要删除的节点id
         */
        void deleteNode(unsigned int id);

        /**
         * pop最小的节点
         * @return T* 返回的最顶部的节点指针
         */
        std::unique_ptr<T> popTopNode();

        /**
         * 获取最顶部的节点
         * @return T 最顶部的节点指针
         */
        std::unique_ptr<T> getTopNode();

        /**
         * 删除顶部的节点
         *
         */
        void deleteTopNode();
};

template<typename T>
SortedHeap<T>::SortedHeap(std::function<bool(T&,T&)> cmp){
    this->cmp = cmp;
    this->autoIncrementId = 0;
}

template<typename T>
void SortedHeap<T>::swap(HeapNode& t1,HeapNode& t2){
    HeapNode tmp = t1;
    t1 = t2;
    t2 = tmp;
}

template<typename T>
void SortedHeap<T>::adjustAfterInsert(){
    int last = this->heap.size() - 1;
    int flag = true;
    //从插入的节点位置开始向上调整
    while(last > 0 && flag){
        if(this->cmp(this->heap[last].obj,this->heap[(last - 1) / 2].obj)){
            this->swap(this->heap[(last - 1) / 2],this->heap[last]);
        }else{
            //不需要调整了
            flag = false;
        }
        last = (last - 1) / 2;
    }
}

template<typename T>
void SortedHeap<T>::adjustAfterDelete(int pos){
    //从pos位置开始向下调整
    int last = this->heap.size() - 1;
    if(last == 0)
        return;     //最后一个不需要调整
    bool flag = true;   //标记是否需要调整
    while(pos <= (last - 1) / 2 && flag){
        //一直调整到最后一个非叶子结点
        int topNum = 0;     //记录最小的结点编号

          //(pos + 1) * 2 - 1是左孩子，pos是父
        if(this->cmp(this->heap[(pos + 1) * 2 - 1].obj,this->heap[pos].obj)){
            topNum = (pos + 1) * 2 - 1;
        }else{
            topNum = pos;
        }

        if((pos + 1) * 2 <= last){
            //如果存在右结点
            if(this->cmp(this->heap[(pos + 1) * 2].obj,this->heap[topNum].obj)){
                topNum = (pos + 1) * 2;
            }
        }

        //看看topNum是不是自己
        if(pos == topNum){
            //是自己就不用调整了
            flag = false;
        }else{
            //交换
            this->swap(this->heap[pos],this->heap[topNum]);
        }
        pos = topNum;
    }
}


template<typename T>
void SortedHeap<T>::deleteNodeByPos(const unsigned int pos){
    unsigned int last = this->heap.size() - 1;
    if(pos > last){
        return;
    }
    std::lock(mu1,mu2);             //上锁
    std::lock_guard<std::mutex> locker1(mu1,std::adopt_lock);
    std::lock_guard<std::mutex> locker2(mu2,std::adopt_lock);
    //与最后一个交换
    swap(this->heap[pos],this->heap[last]);
    //删除最后一个
    this->heap.pop_back();      

    this->adjustAfterDelete(pos);
}



template<typename T>
unsigned int SortedHeap<T>::insertNode(T& node){
    HeapNode hNode(this->autoIncrementId++,node);
    std::lock(mu1,mu2);             //上锁
    std::lock_guard<std::mutex> locker1(mu1,std::adopt_lock);
    std::lock_guard<std::mutex> locker2(mu2,std::adopt_lock);
    this->heap.push_back(hNode);     //先将node放在最后一位
    if(this->heap.size() != 1){
        //如果大小不等于1，则在新增节点后调整
        this->adjustAfterInsert();
    }
    return this->autoIncrementId - 1;
}


template<typename T>
void SortedHeap<T>::deleteNode(unsigned int id){
    for(unsigned int i = 0; i < this->heap.size(); i++){
        if(heap[i].id == id){
            //找到了id
            this->deleteNodeByPos(i);
            break;
        }
    }


}

template<typename T>
std::unique_ptr<T> SortedHeap<T>::popTopNode(){
    if(this->heap.size() != 0){
        std::unique_ptr<T> top(new T(this->heap[0].obj));
        this->deleteNodeByPos(0);
        return top;
    }else{
        std::unique_ptr<T> p = nullptr;
        return p;
    }
}

template<typename T>
std::unique_ptr<T> SortedHeap<T>::getTopNode(){
    if(this->heap.size() != 0){
        std::unique_ptr<T> top(new T(this->heap[0].obj));
        return top;
    }else{
        std::unique_ptr<T> p = nullptr;
        return p;
    }
}

template<typename T>
void SortedHeap<T>::deleteTopNode(){
   if(this->heap.size() != 0){
        this->deleteNodeByPos(0);
    }
}

#endif

5.2 定时器的实现

Timer.h

/**
 * 定时器的实现
 * 支持int setTimer(T interval,function action):设置一个定时器，指定间隔interval和回调函数action,返回定时器id
 * 支持void deleteTimer(int timerId):删除一个定时器
 * 数据结构:最小堆模型，按照定时器触发的时间排序
 * author:jiangpengfei
 * date:2017-05-09
 */
#ifndef TIMER_H
#define TIMER_H
#include <iostream>
#include <chrono>
#include <functional>
#include <thread>
#include <memory>
#include "SortedHeap.hpp"

class Timer{
    private:
        std::chrono::milliseconds tick;
        double timeline;     //当前时间线,long double的字节数为12
        bool isStart;        //标志当前定时器的启动状态
        struct SchedulerEvent{
          unsigned int id;                   //定时事件的唯一标示id
          double interval;                   //事件的触发间隔，在重复事件中会用到这个属性
          double deadline;                   //定时事件的触发时间
          std::function<void()> action;      //触发的事件
          bool isRepeat;                     //是否是重复执行事件
          SchedulerEvent( double interval, double timeline,std::function<void()> action,bool isRepeat){
              this->interval = interval;
              this->deadline = interval + timeline;
              this->action = action;
              this->isRepeat = isRepeat;
          }
        };

        SortedHeap<SchedulerEvent> eventQueue;

        /**
         * 执行到达期限的定时器
         */
        void loopForExecute();

        //私有的构造函数
        Timer(std::chrono::milliseconds tick):eventQueue(
            [](SchedulerEvent& a,SchedulerEvent& b){
                return a.deadline < b.deadline;
            }
        ){
            this->timeline = 0;
            this->tick = tick;
            this->isStart = false;
        }

    public:

        //单例模式
        static Timer* getInstance(std::chrono::milliseconds tick){
            static Timer timer(tick);
            return &timer;
        }

        /**
         * 设置定时器
         * @param interval 定时间隔
         * @param action 定时执行的动作
         * @param isRepeat 是否重复执行,默认不重复执行
         * @return unsigned int 定时器的id,可以根据这个id执行删除操作
         */
        unsigned int addEvent(double interval,std::function<void()> action,bool isRepeat = false);

        /**
         * 删除定时器
         * @param timerId 定时器id
         *
         */
        void deleteEvent(unsigned int timerId);

        /**
         * 同步执行启动定时器
         */
         void syncStart();

         /**
         * 异步执行启动定时器
         */
         void asyncStart();

};


#endif

Timer.cpp

#include "src/core/util/Timer.h"

unsigned int Timer::addEvent(double interval,std::function<void()> action,bool isRepeat){
    SchedulerEvent event(interval,this->timeline,action,isRepeat);
    return this->eventQueue.insertNode(event);
}

void Timer::deleteEvent(unsigned int timerId){
    this->eventQueue.deleteNode(timerId);
}

void Timer::loopForExecute(){
    std::unique_ptr<SchedulerEvent> top = this->eventQueue.getTopNode();
    while(top != nullptr && top->deadline <= this->timeline){
        //如果已经到了执行的时间,新开一个子线程执行任务
        std::thread t(top->action);
        t.detach();    //子线程分离

        if(top->isRepeat){
            //如果是重复事件,则重新添加
            this->addEvent(top->interval,top->action,top->isRepeat);
        }

        //从堆中删除
        this->eventQueue.deleteTopNode();
        top = this->eventQueue.getTopNode();
    }
    //执行一次后等待一个周期
    std::this_thread::sleep_for(this->tick);
    //周期增1
    this->timeline++;
}

void Timer::asyncStart(){
    if(!this->isStart){
        std::thread daemon_thread(&Timer::syncStart,this);
        daemon_thread.detach();     //从当前主线程分离
    }
}

void Timer::syncStart(){
    if(!this->isStart){
        while(1)
            this->loopForExecute();
    }
}

测试执行的代码

#include <iostream>
#include <chrono>
#include <ctime>
#include <iomanip>
#include <string>
#include <functional>
#include <thread>
#include <memory>
#include <fstream>
#include "src/core/util/Timer.h"

void myprint(std::string msg){
    std::ofstream of("timer.txt", std::ios::app);
    std::thread::id this_id = std::this_thread::get_id();
    auto t = std::chrono::system_clock::to_time_t(std::chrono::system_clock::now());
    of << "From Thread " << this_id << "at time " << std::put_time(std::localtime(&t), "%Y-%m-%d %H.%M.%S") << ":" << msg << std::endl;
}

int main(){
    std::chrono::milliseconds tick(2000);       //1000毫秒作为一个周期
    Timer* timer = Timer::getInstance(tick);
    std::function<void()> f1 = std::bind(myprint,"第一个加入,10tick后执行");
    std::function<void()> f2 = std::bind(myprint,"第二个加入，被删除不执行");
    std::function<void()> f3 = std::bind(myprint,"第三个加入，每5tick重复执行");
    std::function<void()> f4 = std::bind(myprint,"第四个加入，5tick后执行");
    std::function<void()> f5 = std::bind(myprint,"第五个加入，5tick后执行");
    std::function<void()> f6 = std::bind(myprint,"第六个加入，5tick后执行");
    std::function<void()> f7 = std::bind(myprint,"第七个加入，5tick后执行");
    std::function<void()> f8 = std::bind(myprint,"第八个加入，5tick后执行");
    std::function<void()> f9 = std::bind(myprint,"第九个加入，5tick后执行");
    std::function<void()> f10 = std::bind(myprint,"第十个加入，5tick后执行");
    std::function<void()> f11 = std::bind(myprint,"第十一个加入，15tick后执行");
    std::function<void()> f12 = std::bind(myprint,"第十二个在执行后加入，20tick+5s后执行");

    timer->addEvent(10,f1);
    int id = timer->addEvent(11,f2);
    timer->addEvent(5,f3,true);
    timer->addEvent(5,f4);
    timer->addEvent(5,f5);
    timer->addEvent(5,f6);   
    timer->addEvent(5,f7);
    timer->addEvent(5,f8);
    timer->addEvent(5,f9);
    timer->addEvent(5,f10);
    timer->addEvent(15,f11);

    timer->deleteEvent(id);

    myprint("线程开始启动,每tick是2秒");

    //异步执行，程序退出后计时器也会终止，因此在下面使用while循环保证程序不会退出
    timer->asyncStart();
    //timer->syncStart();


    //休眠5秒钟
    std::this_thread::sleep_for(std::chrono::seconds(5));   
    //应该在大概20*tick+5秒后执行,
    //TODO 执行后加入的定时器不对
    timer->addEvent(20,f12);

    getchar();

    return 0;
}