Linux高性能服务器编程 学习笔记 第十一章 定时器

网络程序需要处理定时事件，如定期检测一个客户连接的活动状态。服务器进程通常管理着众多定时事件，有效地组织这些定时事件，使其在预期的时间被触发且不影响服务器的主要逻辑，对于服务器的性能有至关重要的影响。为此，我们要将每个定时事件分别封装成定时器，并使用某种容器类数据结构，如链表、排序链表、时间轮，将所有定时器串联起来，以实现对定时事件的统一管理。本章讨论两种高效的管理定时器的容器：时间轮和时间堆。

定时指一段时间后触发某段代码的机制，我们可以在这段代码中依次处理所有到期的定时器，即定时机制是定时器得以被处理的原动力。Linux提供三种定时方法：
1.socket套接字选项SO_RCVTIMEO和SO_SNDTIMEO。

2.SIGALRM信号。

3.IO复用系统调用的超时参数。

SO_RCVTIMEO和SO_SNDTIMEO分别用来设置socket接收和发送数据的超时时间。作者接下来在书里说，这两个socket选项只对socket专用的数据接收和发送系统调用有效，作者给出的专用系统调用为send、sendmsg、recv、recvmsg、accept、connect，但Linux的man page中显示它们适用于所有执行套接字IO操作的系统调用：


由上表，我们可以根据系统调用的返回值和errno来判断超时时间是否已到，进而决定是否开始处理定时任务，下面以connect函数为例，说明SO_SNDTIMEO套接字选项如何来定时：

#include <sys/types.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <arpa/inet.h>
#include <stdlib.h>
#include <assert.h>
#include <stdio.h>
#include <errno.h>
#include <fcntl.h>
#include <unistd.h>
#include <string.h>
#include <libgen.h>// 执行超时connect
int timeout_connect(const char *ip, int port, int time) {int ret = 0;struct sockaddr_in address;bzero(&address, sizeof(address));address.sin_family = AF_INET;inet_pton(AF_INET, ip, &address.sin_addr);address.sin_port = htons(port);int sockfd = socket(PF_INET, SOCK_STREAM, 0);assert(sockfd >= 0);// SO_RCVTIMEO和SO_SNDTIMEO套接字选项对应的值类型为timeval，这和select函数的超时参数类型相同struct timeval timeout;timeout.tv_sec = time;timeout.tv_usec = 0;socklen_t len = sizeof(timeout);ret = setsockopt(sockfd, SOL_SOCKET, SO_SNDTIMEO, &timeout, len);assert(ret != -1);ret = connect(sockfd, (struct sockaddr *)&address, sizeof(address));if (ret == -1) {// 超时对应的错误号是EINPROGRESS，此时就可执行定时任务了if (errno == EINPROGRESS) {printf("conencting timeout, process timeout logic\n");return -1;}printf("error occur when connecting to server\n");return -1;}return sockfd;
}int main(int argc, char *argv[]) {if (argc != 3) {printf("usage: %s ip_address port_number\n", basename(argv[0]));return 1;}const char *ip = argv[1];int port = atoi(argv[2]);int sockfd = timeout_connect(ip, port, 10);if (sockfd < 0) {return 1;}return 0;
}

由alarm和setitimer（可设置周期性触发的定时器）函数设置的闹钟一旦超时，将触发SIGALRM信号，我们可利用该信号的信号处理函数来处理定时任务，但如果要处理多个定时任务，我们就需要不断触发SIGALRM信号。一般，SIGALRM信号按固定的频率生成，即由alarm或setitimer函数设置的定时周期T保持不变，如果某个定时任务的超时时间不是T的整数倍，那么它实际被执行的时间和预期的时间将略有偏差，因此定时周期T反映了定时的精度。

定时器通常至少要包含两个成员：一个超时时间（相对时间或绝对时间）和一个任务回调函数。有时还可能包含回调函数被执行时需要传入的参数，以及是否重启定时器等信息。如果使用链表作为容器来串联所有定时器，则每个定时器还要包含指向下一定时器的指针成员，如果链表是双向的，则每个定时器还需要包含指向前一个定时器的指针成员。

以下代码实现了简单的升序定时器链表，其中的定时器按超时时间做升序排序：

#ifndef LST_TIMER
#define LST_TIMER#include <time.h>
#include <netinet/in.h>
#include <stdio.h>#define BUFFER_SIZE 64// 前向声明，我们需要在client_data结构中定义该结构的指针类型
class util_timer;// 用户数据结构
struct client_data {// 客户socket地址sockaddr_in address;// socket文件描述符int sockfd;// 读缓冲区char buf[BUFFER_SIZE];// 定时器util_timer *timer;
};// 定时器类
class util_timer {
public:util_timer() : prev(NULL), next(NULL) {}// 任务执行时间，UNIX时间戳time_t expire;// 任务回调函数void (*cb_func)(client_data *);// 回调函数处理的客户数据，由定时器的执行者传给回调函数client_data *user_data;// 指向前一个定时器util_timer *prev;// 指向后一个定时器util_timer *next;
};// 定时器链表，它是一个升序、双向链表，且有头节点和尾节点
class sort_timer_lst {
public:sort_timer_lst() : head(NULL), tail(NULL) {}// 链表被删除时，删除其中所有定时器~sort_timer_lst() {util_timer *tmp = head;while (tmp) {head = tmp->next;delete tmp;tmp = head;}}// 将目标定时器timer参数添加到链表中void add_timer(util_timer *timer) {if (!timer) {return;}if (!head) {head = tail = timer;return;}// 如果timer中的执行时间小于链表中所有定时器的超时时间，则将其放在链表头部if (timer->expire < head->expire) {timer->next = head;head->prev = timer;head = timer;return;}// 否则调用重载函数add_timer(util_timer, util_timer)将其放到链表中合适的位置，以保证链表的升序特性add_timer(timer, head);}// 调整定时任务的执行时间，本函数只处理执行时间延后的情况，即将该定时器向链表尾部移动void adjust_timer(util_timer *timer) {if (!timer) {return;}util_timer *tmp = timer->next;// 如果被调整的目标定时器在链表尾，或该定时器的超时值仍小于下一个定时器的超时值，则不用调整if (!tmp || (timer->expire < tmp->expire)) {return;}// 如果目标定时器在链表头，则将该定时器从链表中取出并重新插入链表if (timer == head) {head = head->next;head->prev = NULL;timer->next = NULL;add_timer(timer, head);// 如果目标定时器不是链表头节点，则将该定时器从链表中取出，然后插入其原来所在位置之后的链表中} else {timer->prev->next = timer->next;timer->next->prev = timer->prev;add_timer(timer, timer->next);}}// 将目标定时器timer从链表中删除void del_timer(util_timer *timer) {if (!timer) {return;}// 当链表中只有要删除的那个定时器时if ((timer == head) && (timer == tail)) {delete timer;head = NULL;tail = NULL;return;}// 如果链表中至少有两个定时器，且目标定时器时链表头节点，则将链表的头节点重置为原头节点的下一个节点if (timer == head) {head = head->next;head->prev = NULL;delete timer;return;}// 如果链表中至少有两个定时器，且目标定时器时链表尾节点，则将链表的尾节点重置为原尾节点的前一个节点if (timer == tail) {tail = tail->prev;tail->next = NULL;delete timer;return;}// 如果，目标定时器位于链表中间，则把它前后的定时器串联起来，然后删除目标定时器timer->prev->next = timer->next;timer->next->prev = timer->prev;delete timer;}// SIGALRM信号每次触发就在其信号处理函数（如果使用统一事件源，则是主函数）中执行一次tick函数，处理链表上的到期任务void tick() {if (!head) {return;}printf("timer tick\n");// 获得系统当前UNIX时间戳time_t cur = time(NULL);util_timer *tmp = head;// 从头节点开始依次处理每个定时器，直到遇到一个尚未到期的定时器while (tmp) {// 每个定时器都使用绝对时间作为超时值，因此我们可把定时器的超时值和系统当前时间作比较if (cur < tmp->expire) {break;}// 调用定时器的回调函数，以执行定时任务tmp->cb_func(tmp->user_data);// 执行完定时器中的任务后，将其从链表中删除，并重置链表头节点head = tmp->next;if (head) {head->prev = NULL;}delete tmp;tmp = head;}}private:// 一个重载的辅助函数，它被公有的add_timer和adjust_timer函数调用// 该函数将目标定时器timer参数添加到节点lst_head参数后的链表中void add_timer(util_timer *timer, util_timer *lst_head) {util_timer *prev = lst_head;util_timer *tmp = prev->next;// 遍历lst_head节点后的部分链表，直到找到一个超时时间大于目标定时器超时时间的节点，并将目标定时器插入该节点前while (tmp) {if (timer->expire < tmp->expire) {prev->next = timer;timer->next = tmp;tmp->prev = timer;timer->prev = prev;break;}prev = tmp;tmp = tmp->next;}// 如果遍历完lst_head节点后的链表，仍未找到超时时间大于目标定时器的超时时间的节点，则将目标定时器作为链表尾if (!tmp) {prev->next = timer;timer->prev = prev;timer->next = NULL;tail = timer;}}util_timer *head;util_timer *tail;
};
#endif

sort_timer_lst类是一个升序链表，其核心函数tick相当于心博函数，它每隔一段固定的时间执行一次，以检测并处理到期的任务。判断定时任务到期的依据是定时器的expire值小于当前的系统时间，以检测并处理到期的任务。从执行效率来看，如果链表中有n个定时器，则添加定时器的时间复杂度是O（n），删除定时器的时间复杂度是O（1），执行定时任务的时间复杂度平均是O（1）。

现在考虑以上升序定时器链表的实际应用，处理非活动连接。服务器进程通常要定期处理非活动连接，可这样处理非活动的连接：给客户端发一个重连请求，或关闭该连接，或者其他。Linux在内核中提供了对连接是否处于活动状态的定期检查机制，我们可通过socket选项KEEPALIVE来激活它。我们在应用层实现类似KEEPALIVE的机制，以管理所有长时间处于非活动状态的连接，如以下代码利用alarm函数周期性地触发SIGALRM信号，该信号的信号处理函数利用管道通知主循环执行定时器链表上的定时任务——关闭非活动的连接：

#include <sys/types.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <arpa/inet.h>
#include <assert.h>
#include <stdio.h>
#include <signal.h>
#include <unistd.h>
#include <errno.h>
#include <string.h>
#include <fcntl.h>
#include <stdlib.h>
#include <sys/epoll.h>
#include <pthread.h>
#include <libgen.h>#include "lst_timer.h"#define FD_LIMIT 65535
#define MAX_EVENT_NUMBER 1024
#define TIMESLOT 5static int pipefd[2];
// 用升序链表来管理定时器
static sort_timer_lst timer_lst;
static int epollfd = 0;int setnonblocking(int fd) {int old_option = fcntl(fd, F_GETFL);int new_option = old_option | O_NONBLOCK;fcntl(fd, F_SETFL, new_option);return old_option;
}void addfd(int epollfd, int fd) {epoll_event event;event.data.fd = fd;event.events = EPOLLIN | EPOLLET;epoll_ctl(epollfd, EPOLL_CTL_ADD, fd, &event);setnonblocking(fd);
}void sig_handler(int sig) {int save_errno = errno;int msg = sig;// 此处还是老bug，没有考虑字节序就发送了int的低地址的1字节send(pipefd[1], (char *)&msg, 1, 0);errno = save_errno;
}void addsig(int sig) {struct sigaction sa;memset(&sa, '\0', sizeof(sa));sa.sa_handler = sig_handler;sa.sa_flags |= SA_RESTART;sigfillset(&sa.sa_mask);assert(sigaction(sig, &sa, NULL) != -1);
}void timer_handler() {// 处理定时任务timer_lst.tick();// 由于alarm函数只会引起一次SIGALRM信号，因此重新定时，以不断触发SIGALRM信号alarm(TIMESLOT);
}// 定时器回调函数，它删除非活动连接socket上的注册事件，并关闭之
void cb_func(client_data *user_data) {epoll_ctl(epollfd, EPOLL_CTL_DEL, user_data->sockfd, 0);assert(user_data);close(user_data->sockfd);printf("close fd %d\n", user_data->sockfd);
}int main(int argc, char *argv[]) {if (argc != 3) {printf("usage: %s ip_address port_number\n", basename(argv[0]));return 1;}const char *ip = argv[1];int port = atoi(argv[2]);int ret = 0;struct sockaddr_in address;bzero(&address, sizeof(address));address.sin_family = AF_INET;inet_pton(AF_INET, ip, &address.sin_addr);address.sin_port = htons(port);int listenfd = socket(PF_INET, SOCK_STREAM, 0);assert(listenfd >= 0);ret = bind(listenfd, (struct sockaddr *)&address, sizeof(address));assert(ret != -1);ret = listen(listenfd, 5);assert(ret != -1);epoll_event events[MAX_EVENT_NUMBER];int epollfd = epoll_create(5);assert(epollfd != -1);addfd(epollfd, listenfd);ret = socketpair(PF_UNIX, SOCK_STREAM, 0, pipefd);assert(ret != -1);setnonblocking(pipefd[1]);addfd(epollfd, pipefd[0]);// 设置信号处理函数addsig(SIGALRM);addsig(SIGTERM);bool stop_server = false;// 直接初始化FD_LIMIT个client_data对象，其数组索引是文件描述符client_data *users = new client_data[FD_LIMIT];bool timeout = false;// 定时alarm(TIMESLOT);while (!stop_server) {int number = epoll_wait(epollfd, events, MAX_EVENT_NUMBER, -1);if ((number < 0) && (errno != EINTR)) {printf("epoll failure\n");break;}for (int i = 0; i < number; ++i) {int sockfd = events[i].data.fd;// 处理新到的客户连接if (sockfd == listenfd) {struct sockaddr_in client_address;socklen_t client_addrlength = sizeof(client_address);int connfd = accept(listenfd, (struct sockaddr *)&client_address, &client_addrlength);addfd(epollfd, connfd);users[connfd].address = client_address;users[connfd].sockfd = connfd;// 创建一个定时器，设置其回调函数和超时时间，然后绑定定时器和用户数据，并将定时器添加到timer_lst中util_timer *timer = new util_timer;timer->user_data = &users[connfd];timer->cb_func = cb_func;time_t cur = time(NULL);timer->expire = cur + 3 * TIMESLOT;users[connfd].timer = timer;timer_lst.add_timer(timer);// 处理信号} else if ((sockfd == pipefd[0]) && (events[i].events & EPOLLIN)) {int sig;char signals[1024];ret = recv(pipefd[0], signals, sizeof(signals), 0);if (ret == -1) {// handle the errorcontinue;} else if (ret == 0) {continue;} else {for (int i = 0; i < ret; ++i) {switch (signals[i]) {case SIGALRM:// 先标记为有定时任务，因为定时任务优先级比IO事件低，我们优先处理其他更重要的任务timeout = true;break;case SIGTERM:stop_server = true;break;}}}// 从客户连接上接收到数据} else if (events[i].events & EPOLLIN) {memset(users[sockfd].buf, '\0', BUFFER_SIZE);ret = recv(sockfd, users[sockfd].buf, BUFFER_SIZE - 1, 0);printf("get %d bytes of client data %s from %d\n", ret, users[sockfd].buf, sockfd);util_timer *timer = users[sockfd].timer;if (ret < 0) {// 如果发生读错误，则关闭连接，并移除对应的定时器if (errno != EAGAIN) {cb_func(&users[sockfd]);if (timer) {timer_lst.del_timer(timer);}}} else if (ret == 0) {// 如果对方关闭连接，则我们也关闭连接，并移除对应的定时器cb_func(&users[sockfd]);if (timer) {timer_lst.del_timer(timer);}} else {// 如果客户连接上读到了数据，则调整该连接对应的定时器，以延迟该连接被关闭的时间if (timer) {time_t cur = time(NULL);timer->expire = cur + 3 * TIMESLOT;printf("adjust timer once\n");timer_lst.adjust_timer(timer);}}}}// 最后处理定时事件，因为IO事件的优先级更高，但这样会导致定时任务不能精确按预期的时间执行if (timeout) {timer_handler();timeout = false;}}close(listenfd);close(pipefd[1]);close(pipefd[0]);delete[] users;return 0;
}

Linux下的3组IO复用系统调用都带有超时参数，因此它们不仅能统一处理信号（通过管道在信号处理函数中通知主进程）和IO事件，也能统一处理定时事件，但由于IO复用系统调用可能在超时时间到期前就返回（有IO事件发生），所以如果我们要利用它们来定时，就需要不断更新定时参数以反映剩余的时间，如下代码所示：

#define TIMEOUT 5000int timeout = TIMEOUT;
time_t start = time(NULL);
time_t end = time(NULL);
while (1) {printf("the timeout is now %d mil-seconds\n", timeout);start = time(NULL);int number = epoll_wait(epollfd, events, MAX_EVENT_NUMBER, timeout);if ((number < 0) && (errno != EINTR)) {printf("epoll failure\n");break;}// 如果epoll_wait函数返回0，说明超时时间到，此时可处理定时任务，并重置定时时间if (number == 0) {// 处理定时任务timeout = TIMEOUT;continue;}// 到此，epoll_wait函数的返回值大于0,end = time(NULL);// 更新timeout的值为减去本次epoll_wait调用的持续时间timeout -= (end - start) * 1000;// 重新计算后的timeout值可能是0，说明本次epoll_wait调用返回时，不仅有文件描述符就绪，且其超时时间也刚好到达// 此时我们要处理定时任务，并充值定时时间if (timeout <= 0) {// 处理定时任务timeout = TIMEOUT;}// handle connections
}

基于排序链表的定时器存在一个问题：添加定时器的效率偏低。下面讨论的时间轮解决了这个问题，一种简单的时间轮如下图：

上图中，时间轮内部的实线指针指向轮子上的一个槽（slot），它以恒定的速度顺时针转动，每转动一步就指向下一个槽（虚线指针所指的槽），每次转动称为一个滴答（tick），一个滴答的时间称为时间轮的槽间隔si（slot interval），它实际上就是心博时间。上图中的时间轮共有N个槽，因此它转动一周的时间是Nsi。每个槽指向一条定时器链表，每条链表上的定时器具有相同的特征：它们的定时事件相差Nsi的整数倍，时间轮正是利用这个关系将定时器散列到不同的链表中。假如现在指针指向槽cs，我们要添加一个定时事件为ti的定时器，则该定时器将被插入槽ts（timer slot）对应的链表中：

基于排序链表的定时器使用唯一的一条链表来管理所有定时器，所以插入操作的效率随着定时器数目的增多而降低，而时间轮使用哈希表的思想，将定时器散列到不同的链表上，这样每条链表上的定时器数目都将明显少于原来的排序链表上的定时器数目，插入操作的效率基本不受定时器数目的影响。

对时间轮而言，要想提高定时精度，就要使si足够小，要提高执行效率，就要求N足够大（N越大，散列冲突的概率就越小）。

以下代码描述了一种简单的时间轮，因为它只有一个轮子，而复杂的时间轮可能有多个轮子，不同的轮子有不同的粒度，相邻的两个轮子，精度高的转一圈，精度低的仅仅往前移动一槽，就像水表一样：

#ifndef TIME_WHEEL_TIMER
#define TIME_WHEEL_THMER#include <time.h>
#include <netinet/in.h>
#include <stdio.h>#define BUFFER_SIZE 64class tw_timer;// 用户数据，包含socket和定时器
struct client_data {sockaddr_in address;int sockfd;char buf[BUFFER_SIZE];tw_timer *timer;
};// 定时器类
class tw_timer {
public:tw_timer(int rot, int ts) : next(NULL), prev(NULL), rotation(rot), time_slot(ts) { }// 定时器在时间轮转多少圈后生效int rotation;// 定时器属于时间轮上的哪个槽（对应的链表）int time_slot;// 定时器回调函数void (*cb_func)(client_data *);// 客户数据client_data *user_data;// 指向下一个定时器tw_timer *next;// 指向前一个定时器tw_timer *prev;
};class time_wheel {
public:time_wheel() : cur_slot(0) {for (int i = 0; i < N; ++i) {// 初始化每个槽的头节点slots[i] = NULL;}}~time_wheel() {// 销毁每个槽中的所有定时器for (int i = 0; i < N; ++i) {tw_timer *tmp = slots[i];while (tmp) {slots[i] = tmp->next;delete tmp;tmp = slots[i];}}}// 根据定时值timeout参数创建一个定时器，并将它插入合适的槽中tw_timer *add_timer(int timeout) {if (timeout < 0) {return NULL;}int ticks = 0;// 计算待插入定时器的超时值在多少滴答后被触发// 如果待插入定时器的超时值小于时间轮的槽间隔，就将其向上折合成1if (timeout < SI) {ticks = 1;// 否则将其向下折合成timeout/SI} else {ticks = timeout / SI;}// 计算待插入的定时器在时间轮转动多少圈后触发int rotation = ticks / N;// 计算待插入的定时器应被插入哪个槽中int ts = (cur_slot + (ticks % N)) % N;// 创建新定时器，它在时间轮转动rotation圈后触发，且位于第ts个槽上tw_timer *timer = new tw_timer(rotation, ts);// 如果第ts个槽为空，则将新建的定时器作为该槽的头节点if (!slots[ts]) {printf("add timer, rotation is %d, ts is %d, cur_slot is %d\n", rotation, ts, cur_slot);slots[ts] = timer;// 否则，将定时器插入第ts个槽中，也作为头节点} else {timer->next = slots[ts];slots[ts]->prev = timer;slots[ts] = timer;}return timer;}// 删除目标定时器void del_timer(tw_timer *timer) {if (!timer) {return;}int ts = timer->time_slot;// slots[ts]是目标定时器时，说明目标定时器是该槽的头节点，此时需要重置第ts个槽的头节点if (timer == slots[ts]) {slots[ts] = slots[ts]->next;if (slots[ts]) {slots[ts]->prev = NULL;}delete timer;} else {timer->prev->next = timer->next;if (timer->next) {timer->next->prev = timer->prev;}delete timer;}}// SI时间到后，调用该函数，时间轮向前滚动一个槽的间隔void tick() {// 取得时间轮上当前槽的头节点tw_timer *tmp = slots[cur_slot];printf("current slot is %d\n", cur_slot);while (tmp) {printf("tick the timer once\n");// 如果定时器的rotation值大于0，则它在这一轮不起作用if (tmp->rotation > 0) {--tmp->rotation;tmp = tmp->next;// 否则说明定时器已到期，于是执行定时任务，然后删除该定时器} else {tmp->cb_func(tmp->user_data);if (tmp == slots[cur_slot]) {printf("delete header in cur_slot\n");slots[cur_slot] = tmp->next;delete tmp;if (slots[cur_slot]) {slots[cur_slot]->prev = NULL;}tmp = slots[cur_slot];} else {tmp->prev->next = tmp->next;if (tmp->next) {tmp->next->prev = tmp->prev;}tw_timer *tmp2 = tmp->next;delete tmp;tmp = tmp2;}}}// 更新时间轮的当前值，以反映时间轮转动cur_slot = ++cur_slot % N;}private:// 时间轮上的槽数static const int N = 60;// 每1秒时间轮转动一次，即槽间隔为1秒static const int SI = 1;// 时间轮上的槽，其中每个元素指向一个定时器的链表，链表无序tw_timer *slots[N];// 时间轮的当前槽int cur_slot;
};#endif

可见，对时间轮而言，如果一共有n个定时器，则添加一个定时器的时间复杂度为O（1）；删除一个定时器的时间复杂度平均也是O（1），但最坏情况下可能所有节点都在一个槽中，此时删除定时器的时间复杂度为O（n）；执行一个定时器的时间复杂度是O（n），但如果分布很平均，则时间复杂度为O（n/N），N是时间轮的槽数。实际上执行一个定时器任务的效率要比O（n）好得多，因为时间轮将所有定时器散列到了不同的链表上，时间轮的槽越多，等价于散列表的入口（entry）越多，从而每条链表上的定时器数量越少。此外，以上代码中只使用了1个时间轮，当使用多个轮子来实现时间轮时，执行一个定时器任务的时间复杂度将接近O（1）。

以上讨论的定时方案都是以固定频率调用心博函数tick，并依次检测到期的定时器，然后执行定时器上的回调函数。设计定时器的另一种思路是：将所有定时器中超时时间最小的定时器的超时值作为心博间隔，这样，一旦心博函数tick被调用，超时时间最小的定时器必然到期，我们就可在tick函数中处理该定时器，然后，再次从剩余定时器中找出超时时间最小的一个，并将这段最小时间设为下一次心博间隔。

最小堆很适合这种定时方案，最小堆是每个节点的值都小于或等于其子节点的值的完全二叉树，一个最小堆的例子：

树的基本操作是插入节点和删除节点，对最小堆而言，为了将一个元素X插入最小堆，我们可以在树的下一个空闲位置创建一个空穴，如果X可以放在空穴中而不破坏堆序，则插入完成，否则执行上虑操作，即交换空穴和它的父节点上的元素，不断执行上述过程，直到X可以被放入空穴，则插入操作完成，例如，我们要往下图最左面所示的最小堆中插入值为14的元素，可按下图步骤完成：

最小堆的删除操作指的是删除其根节点上的元素，且不破坏堆序性质，执行删除操作时，我们需要先在根节点处创建一个空穴，由于堆现在少了一个元素，因此我们把堆的最后一个元素X移动到该堆的某个地方，如果X可以被放入空穴（当前是根节点），则删除操作完成，否则执行下虑操作，即交换空穴和它的两个子节点中较小者，不断进行该过程，直到X可以被放入空穴。例如，我们对图11-2中的最小堆执行删除操作，以下是删除步骤：

我们可用数组来组织最小堆中的元素，图11-2所示的最小堆可用下图所示数组来表示：

对于数组中任意位置i上的元素，其左儿子节点在位置2i+1上，其右儿子在位置2i+2上，其父节点在[(i-1)/2]上。与用链表来表示堆相比，用数组表示堆不仅节省空间，且更容易实现堆的插入、删除等操作。

假设我们已经有一个包含N个元素的数组，现在把它初始化为一个最小堆，最简单的方法是：初始化一个空堆，然后将数组中的每个元素插入该堆中，但这样做效率偏低，实际上，我们只需对数组中第[(N-1)/2]到0个元素执行下虑操作，即可保证该数组构成一个最小堆，因为对包含N个元素的完全二叉树而言，它具有[(N-1)/2]个非叶子节点，这些非叶子节点正是该完全二叉树的第0到[(N-1)/2]个节点，我们只需确保这些非叶子节点构成的子树具有堆序性质，整个树就具有堆序性质。

我们称用最小堆实现的定时器为时间堆，以下给出一种时间堆的实现，其中最小堆使用数组表示：

#ifndef MIN_HEAP
#define MIN_HEAP#include <iostream>
#include <netinet/in.h>
#include <time.h>
using std::exception;#define BUFFER_SIZE 64// 前向声明
class heap_timer;// 客户数据，绑定socket和定时器
struct client_data {sockaddr_in address;int sockfd;char buf[BUFFER_SIZE];heap_timer *timer;
};// 定时器类
class heap_timer {
public:heap_timer(int delay) {expire = time(NULL) + delay;}// 定时器生效的绝对时间time_t expire;// 定时器的回调函数void (*cb_func)(client_data *);// 用户数据client_data *user_data;
};// 时间堆类
class time_heap {
public:// 构造函数一，初始化一个大小为cap的空堆// throw (std::exception)是异常规范指出函数可能抛出的异常类型，是给调用者的提示而非强制性的规则// 在C++11中，异常规范已经被弃用// 上面作者已经使用了using将std::exception引入了当前作用域，此处可以不加std::前缀time_heap(int cap) throw (std::exception) : capacity(cap), cur_size(0) {// 创建堆数组// 此处代码是错的，new在分配失败时默认会抛出std::bad_alloc异常array = new heap_timer* [capacity];if (!array) {throw std::exception();}for (int i = 0; i < capacity; ++i) {array[i] = NULL;}}// 构造函数二，用已有代码初始化堆time_heap(heap_timer **init_array, int size, int capacity) throw (std::exception) : cur_size(size), capacity(capacity) {if (capacity < size) {throw std::exception();}          // 创建堆数组// 此处代码是错的，new在分配失败时默认会抛出std::bad_alloc异常array = new heap_timer* [capacity];if (!array) {throw std::exception();}for (int i = 0; i < capacity; ++i) {array[i] = NULL;}if (size != 0) {// 初始化堆数组for (int i = 0; i < size; ++i) {array[i] = init_array[i];}// 对数组中第[(cur_size - 1) / 2]到0之间的元素执行下虑操作for (int i = (cur_size - 1) / 2; i >= 0; --i) {percolate_down(i);}}}// 销毁时间堆~time_heap() {for (int i = 0; i < cur_size; ++i) {delete array[i];}delete[] array;}// 添加定时器timervoid add_timer(heap_timer *timer) throw (std::exception) {if (!timer) {return;}// 如果当前堆数组容量不足，则将其扩大if (cur_size >= capacity) {resize();}// 新插入了一个元素，当前堆大小加1，hole是新建空穴的位置int hole = cur_size++;int parent = 0;// 对从空穴到根节点路径上的所有节点执行上虑操作for (; hole > 0; hole = parent) {parent = (hole - 1) / 2;if (array[parent]->expire <= timer->expire) {break;}array[hole] = array[parent];}array[hole] = timer;}// 删除定时器void del_timer(heap_timer *timer) {if (!timer) {return;}// 仅仅将目标定时器的回调函数设置为空，即所谓的延迟销毁// 这样节省真正删除该定时器造成的开销，但容易使堆数组膨胀timer->cb_func = NULL;}// 获得堆顶的定时器// const成员函数不会改变对象中的数据成员heap_timer *top() const {if (empty()) {return NULL;}return array[0];}// 删除堆顶的定时器void pop_timer() {if (empty()) {return;}if (array[0]) {delete array[0];// 将原来堆顶元素替换为堆数组中最后一个元素array[0] = array[--cur_size];// 对新的堆顶元素执行下虑操作percolate_down(0);}}// 心博函数void tick() {heap_timer *tmp = array[0];time_t cur = time(NULL);// 循环处理堆中到期的定时器while (!empty()) {if (!tmp) {break;}// 如果堆顶定时器没到期，则退出循环if (tmp->expire > cur) {break;}// 否则就执行堆顶定时器中的任务if (array[0]->cb_func) {array[0]->cb_func(array[0]->user_data);}// 删除堆顶元素pop_timer();tmp = array[0];}}bool empty() const {return cur_size == 0;}private:// 最小堆的下虑操作，它确保数组中以第hole个节点作为根的子树拥有最小堆性质void percolate_down(int hole) {heap_timer *temp = array[hole];int child = 0;for (; (hole * 2 + 1) <= (cur_size - 1); hole = child) {child = hole * 2 + 1;// child < cur_size - 1保证了hole有右子节点if ((child < cur_size - 1) && (array[child + 1]->expire < array[child]->expire)) {++child;}if (array[child]->expire < temp->expire) {array[hole] = array[child];} else {break;}}array[hole] = temp;}// 将堆数组容量扩大一倍void resize() throw (std::exception) {heap_timer **temp = new heap_timer* [2 * capacity];for (int i = 0; i < 2 * capacity; ++i) {temp[i] = NULL;}if (!temp) {throw std::exception();}capacity = 2 * capacity;for (int i = 0; i < cur_size; ++i) {temp[i] = array[i];}delete[] array;array = temp;}// 堆数组heap_timer **array;// 堆数组的容量int capacity;// 堆数组中当前包含元素的个数int cur_size;
};#endif

对时间堆而言，如果有n个定时器，则添加一个定时器的时间复杂度是O（lgn）；删除一个定时器的时间复杂度是O（1）；执行一个定时器的时间复杂度是O（1）。