当前位置：首页 > article >正文

使用epoll监测定时器是否到达指定时间，并执行回调函数

article 2024/12/5 3:25:48

总览：Linux提供了定时器，暴露出来了文件描述符，所以我们使用epoll帮助我们监测，时间到达后，epoll_wait返回，于是我们根据fd，找到对应的回调函数，然后执行。从而达到定时执行函数的目的，后续我们打算将它添加到线程池中。本来我们的线程池如果遇到定时任务是睡眠n秒，然后执行，但是显而易见：这对于线程资源来说就是巨大的浪费。这样问题就得到了改善。

架构：

代码：

mainTest.cpp

// C++ STL
#include <thread>
#include <iostream>
using namespace std;
#include "TimerManage.hpp"
void func(int x)
{
  static int num = 0;
  cout << "func " << x << " num: " << ++num << endl;
}
int main()
{
  myspace::Timer t1;
  myspace::TimerMap tmp;
  t1.init();
  t1.set_timer(std::bind(func, 1), 1250);
  tmp.init(-1);
  tmp.add_timer(t1);

  myspace::Timer t2;
  t2.init();
  t2.set_timer(std::bind(func, 100), 2250);
  tmp.add_timer(t2);
  
  while (1)
  {
    static int seconds = 0;
    std::this_thread::sleep_for(std::chrono::seconds(1));
    cout << "___________________________main while: " << ++seconds << " s" << endl;
  }
  return 0;
}

Timer.hpp

//一个Timer对应一个任务，但是线程池中任务众多，所以我们需要对他做封装，搞出来一个Timer的管理者，所以就有了TimerManager，并且我们还要让epoll参与管理这么多fd（Timer对应一个fd）
#include <functional>
using namespace std;
#ifndef TIMER_HPP
#define TIMER_HPP
namespace myspace
{
  class Timer
  {
  public:
    using TimerCallback = std::function<void(void)>;

  private:
    int m_timerId;
    TimerCallback m_callback;
    int tag; // 1 out; 2 inter;
    bool settimer(size_t interval);

  public:
    Timer();
    ~Timer();
    Timer(const Timer &) = delete;
    Timer &operator=(const Timer &) = delete;
    Timer(Timer &&);
    Timer &operator=(Timer &&);
    bool init();
    bool set_timer(const TimerCallback &cb, size_t interval);
    bool reset_timer(size_t interval);
    void handle_event();
    int get_TimerId() const;
    int close_timer();
  };
}

#endif

Timer.cpp



// Liunx API
#include <sys/timerfd.h>
// C API
#include <unistd.h>
#include <stdio.h>
#include <string.h>
// C++ STL;
#include <iostream>
using namespace std;
#include "Timer.hpp"

namespace myspace
{
  // class Timer
  // using TimerCallback = std::function<void(void)>;
  // int m_timerId;
  // TimerCallback m_callback;
  // int tag; // 1 out; 2 inter;
  bool Timer::settimer(size_t interval) // 1250
  {
    bool ret = true;
    struct itimerspec new_value = {0};
    new_value.it_interval.tv_sec = (interval / 1000); // secodes;
    new_value.it_interval.tv_nsec = (interval % 1000) * 1000 * 1000;
    new_value.it_value = new_value.it_interval;
    if (timerfd_settime(m_timerId, 0, &new_value, nullptr) < 0)
    {
      fprintf(stderr, "failer error : %s \n", strerror(errno));
      ret = false;
    }
    return ret;
  }

  Timer::Timer() : m_timerId(-1), m_callback(nullptr) {}
  Timer::~Timer() { close_timer(); }
  Timer::Timer(Timer &&other)
      : m_timerId(other.m_timerId),
        m_callback(other.m_callback)
  {
    other.m_timerId = -1;
    other.m_callback = nullptr;
  }
  Timer &Timer::operator=(Timer &&other)
  {
    if (this == &other)
      return *this;
    //??
    m_timerId = other.m_timerId;
    m_callback = other.m_callback;
    other.m_timerId = -1;
    other.m_callback = nullptr;
    return *this;
  } //
  bool Timer::init()
  {
    bool ret = true;
    if (m_timerId > 0)
    {
      return ret;
    }
    m_timerId = timerfd_create(CLOCK_MONOTONIC, 0);
    if (m_timerId < 0)
    {
      ret = false;
    }
    return ret;
  }
  bool Timer::set_timer(const TimerCallback &cb, size_t interval)
  {
    bool ret = false;
    if (m_timerId > 0 && settimer(interval))
    {
      m_callback = cb;
      ret = true;
    }
    return ret;
  }
  bool Timer::reset_timer(size_t interval)
  {
    bool ret = false;
    if (m_timerId > 0 && m_callback != nullptr && settimer(interval))
    {
      ret = true;
    }
    return ret;
  }
  void Timer::handle_event()
  {
    uint64_t expire_cnt = 0;
    if (read(m_timerId, &expire_cnt, sizeof(expire_cnt)) != sizeof(expire_cnt)) // 定时器到期时，会在这个文件描述符上产生可读事件。read 用来消耗这个事件，epoll就不管了。这和网络中的epoll一样，你要read或者recv一下。而read的这个值就是到期的次数，：定时器到期一次，返回1，然后你去缓冲区中去消耗这个1

    {
      return;
    }
    cout << "expire_cnt : " << expire_cnt << endl;
    if (m_callback != nullptr)
    {
      m_callback();
    }
  }
  int Timer::get_TimerId() const
  {
    return m_timerId;
  }
  int Timer::close_timer()
  {
    bool ret = false;
    if (m_timerId > 0)
    {
      close(m_timerId);
      m_timerId = -1;
      m_callback = nullptr;
      ret = true;
    }
    return ret;
  }
}

TimerManage.hpp

// ONWER

#include "Timer.hpp"
// LIUNX API
#include <sys/epoll.h>
// C++ STL
#include <thread>
#include <map>
#include <unordered_map>
#include <vector>

#ifndef TIMER_MANAGE_HPP
#define TIMER_MANAGE_HPP

namespace myspace
{
  class TimerMap
  {
  private:
    int m_epollfd;
    std::vector<epoll_event> m_events;
    std::unordered_map<int, myspace::Timer> m_timers;
    bool m_stop; // true stop;
    static const int eventsize = 16;
    std::thread m_workerThread;

  public:
    TimerMap();
    ~TimerMap();
    TimerMap(const TimerMap &) = delete;
    TimerMap &operator=(const TimerMap &) = delete;
    bool init(int timeout);
    bool add_timer(Timer &tv);
    void remove_timer(int fd);
    void loop(int timeout);
    void set_stop();
  };
}

#endif

TimerManage.cpp


#include "TimerManage.hpp"
// C API
#include <string.h>
#include <unistd.h>
//C++ API
#include<iostream>
using namespace std;
namespace myspace
{
  // TimerMap
  // int m_epollfd;
  // std::vector<epoll_event> m_events;
  // std::unordered_map<int, myspace::Timer> m_timers;
  // bool m_stop; // true stop;
  // static const int eventsize = 16;
  // std::thread m_workerThread;
  TimerMap::TimerMap()
      : m_epollfd(-1), m_stop(false)
  {
    m_events.resize(eventsize);
  }
  TimerMap::~TimerMap()
  {
    set_stop();
  }

  bool TimerMap::init(int timeout)
  {
    bool ret = false;
    m_epollfd = epoll_create1(EPOLL_CLOEXEC);
    if (m_epollfd > 0)
    {
      try
      {
        m_workerThread = std::thread(&TimerMap::loop, this, timeout);
        ret = true;
      }
      catch (const std::exception &e)
      { 
        cout << e.what() << '\n';
        close(m_epollfd);
        m_epollfd = -1;
      }
    }
    return ret;
  }

  bool TimerMap::add_timer(Timer &tv)
  {
    struct epoll_event evt;
    evt.data.fd = tv.get_TimerId();
    evt.events = EPOLLIN | EPOLLET;
    if (epoll_ctl(m_epollfd, EPOLL_CTL_ADD, tv.get_TimerId(), &evt) < 0)
    {
      fprintf(stderr, "timer_manager: epoll_ctl_add failed, errno = %s\n",
              strerror(errno));
      return false;
    }
    m_timers[tv.get_TimerId()] = std::move(tv);
    return true;
  }
  void TimerMap::remove_timer(int fd)
  {
    if(m_timers.find(fd)==m_timers.end())
      return;
    epoll_ctl(fd, EPOLL_CTL_DEL, fd, nullptr);
    m_timers.erase(fd);    
  }
  void TimerMap::loop(int timeout)
  {
    while (!m_stop)
    {
      int n = epoll_wait(m_epollfd, m_events.data(), m_events.size(), timeout);//timeout==-1表示阻塞等待
      for (int i = 0; i < n; ++i)
      {
        int fd = m_events[i].data.fd;
        auto it = m_timers.find(fd);
        if (it != m_timers.end())
        {
          Timer &tv = it->second;
          tv.handle_event();
        }
      }
      if (n >= m_events.size())
      {
        m_events.resize(m_events.size() * 2);
      }
    }
  }
  void TimerMap::set_stop()
  {
    m_stop = true;
    if (m_workerThread.joinable())
    {
      m_workerThread.join();
    }
    close(m_epollfd);
    m_epollfd = -1;
  }
}

编译命令：