基于多反应堆的高并发服务器【C/C++/Reactor】（中）处理任务队列中的任务

本文主要是介绍基于多反应堆的高并发服务器【C/C++/Reactor】（中）处理任务队列中的任务，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！

一、处理任务队列中的任务

（1）EventLoop启动 EventLoop初始化和启动

// 启动反应堆模型
int eventLoopRun(struct EventLoop* evLoop) {assert(evLoop != NULL);// 取出事件分发和检测模型struct Dispatcher* dispatcher = evLoop->dispatcher;// 比较线程ID是否正常if(evLoop->threadID != pthread_self()) {return -1;}// 循环进行事件处理while(!evLoop->isQuit) {dispatcher->dispatch(evLoop,2); // 超时时长 2s// 已续写eventLoopProcessTask(evLoop);}return 0;   
}

多加一句eventLoopProcessTask(evLoop); 理由在后文提到！

（2）添加任务到任务队列中（在EventLoop的任务队列中添加新任务）

// 添加任务到任务队列
int eventLoopAddTask(struct EventLoop* evLoop,struct Channel* channel,int type) {// 加锁，保护共享资源pthread_mutex_lock(&evLoop->mutex);// 创建新节点,后添加到任务队列中去struct ChannelElement* node = (struct ChannelElement*)malloc(sizeof(struct ChannelElement));node->channel = channel;node->type = type;node->next = NULL;// 链表为空if(evLoop->head == NULL) {evLoop->head = evLoop->tail = node;}else {evLoop->tail->next = node; // 添加evLoop->tail = node; // 后移}pthread_mutex_unlock(&evLoop->mutex);if(evLoop->threadID == pthread_self()) {// 当前子线程eventLoopProcessTask(evLoop);}else{// 主线程 -- 告诉子线程处理任务队列中的任务// 1.子线程在工作 2.子线程被阻塞了：select、poll、epolltaskWakeup(evLoop);}return 0;
}

小细节：假设说添加任务的是主线程，那么程序就会执行taskWakeup这个函数，主线程执行这个函数，对于子线程来说有两种情况：

第一种情况，它正在干活，对于子线程没有影响，充其量就是它检测的那个集合里边多出来了一个被激活的文件描述符。

第二种情况，如果说此时子线程被select、poll、或epoll_wait阻塞了,调用taskWakeup函数可以解除其阻塞。如果解除阻塞了，我们希望子线程干什么事情呢？

具体流程：因为主线程是在子线程的任务队列里添加了一个任务，先调用taskWakeup函数解除阻塞，让子线程解除阻塞是需要让它去处理任务。接着在eventLoopRun函数中调用eventLoopProcessTask(evLoop);
详解流程：因为这个反应堆模型只要开始运行（eventLoopRun）就会不停的调用dispatch函数，这个dispatch是一个函数指针，底层指向的是poll模型的poll函数，select模型的select函数，epoll模型的epoll_wait函数。如果当前的子线程正在被刚才的提到的这三个函数里边的其中一个阻塞着，此时正好被主线程唤醒了（调用taskWakeup函数）。需要在循环进行事件处理中添加一句eventLoopProcessTask(evLoop);

taskWakeup函数的调用和影响

主线程调用taskWakeup函数时，子线程可能正在被select、poll、或epoll_wait阻塞
主线程在子线程的任务队列中添加任务，因此需要让子线程解除阻塞并处理队列中的任务

总结关于任务队列的处理有两个路径：

第一个路径：子线程往任务队列里边添加一个任务，比如说修改文件描述符里边的事件，肯定是子线程自己修改自己检测的文件描述符的事件，修改完了之后，子线程就直接调用这个函数去处理任务队列里边的任务。
第二个路径：主线程在子线程的任务队列里边添加了一个任务，主线程是处理不了的，并且主线程现在也不知道子线程是在工作还是在阻塞，所以主线程就默认子线程现在正在阻塞，因此主线程就调用了一个唤醒函数（taskWakeup）来解除阻塞，调用这个函数保证子线程肯定是在运行的，而子线程是eventLoopRun函数的dispatch函数的调用位置解除了阻塞，然后调用eventLoopProcessTask(evLoop);

// 启动反应堆模型
int eventLoopRun(struct EventLoop* evLoop) {...// 循环进行事件处理while(!evLoop->isQuit) {dispatcher->dispatch(evLoop,2); // 超时时长 2s// 已续写eventLoopProcessTask(evLoop);}return 0;   
}// 添加任务到任务队列
int eventLoopAddTask(struct EventLoop* evLoop,struct Channel* channel,int type) {...if(evLoop->threadID == pthread_self()) {// 当前子线程eventLoopProcessTask(evLoop);}else{// 主线程 -- 告诉子线程处理任务队列中的任务// 1.子线程在工作 2.子线程被阻塞了：select、poll、epolltaskWakeup(evLoop);}return 0;
}

EventLoop.h

// 处理任务队列中的任务
int eventLoopProcessTask(struct EventLoop* evLoop);// 处理dispatcher中的任务
int eventLoopAdd(struct EventLoop* evLoop,struct Channel* channel);
int eventLoopRemove(struct EventLoop* evLoop,struct Channel* channel);
int eventLoopModify(struct EventLoop* evLoop,struct Channel* channel);

EventLoop.c

eventLoopProcessTask函数的作用：它处理队列中的任务，需要遍历链表并根据type进行对应处理。

在加锁和解锁函数之间遍历链表
需要遍历链表，处理节点，并在处理完后从列表中删除节点
如果当前节点被处理完，需要移动head指针以处理下一个节点
需要定义temp指针来保存head指针指向的地址，并在head后移后释放temp指针指向的节点

// 处理任务队列中的任务
int eventLoopProcessTask(struct EventLoop* evLoop) {pthread_mutex_lock(&evLoop->mutex);// 取出头节点struct ChannelElement* head = evLoop->head;while (head!=NULL) {struct Channel* channel = head->channel;if(head->type == ADD) {// 添加eventLoopAdd(evLoop,channel);}else if(head->type == DELETE) {// 删除eventLoopRemove(evLoop,channel);}else if(head->type == MODIFY) {// 修改eventLoopModify(evLoop,channel);}struct ChannelElement* tmp = head;head = head->next;// 释放节点free(tmp);}evLoop->head = evLoop->tail = NULL;pthread_mutex_unlock(&evLoop->mutex);return 0;
}

注意事项

在写程序时，每个功能应对应一个任务函数，以提高程序逻辑性和可维护性
在处理任务队列时，需要注意线程安全和资源管理问题

// 处理dispatcher中的任务
int eventLoopAdd(struct EventLoop* evLoop,struct Channel* channel) {int fd = channel->fd;struct ChannelMap* channelMap = evLoop->channelMap;if(fd >= channelMap->size) {// 没有足够的空间存储键值对 fd->channel ==> 扩容if(!makeMapRoom(channelMap,fd,sizeof(struct Channel*))) {return -1;}}// 找到fd对应的数组元素位置，并存储if(channelMap->list[fd] == NULL) {channelMap->list[fd] = channel;evLoop->dispatcher->add(channel,evLoop);} return 0;
}int eventLoopRemove(struct EventLoop* evLoop,struct Channel* channel) {int fd = channel->fd;struct ChannelMap* channelMap = evLoop->channelMap;if(fd >= channelMap->size) {return -1;}int ret = evLoop->dispatcher->remove(channel,evLoop);return ret;
}int eventLoopModify(struct EventLoop* evLoop,struct Channel* channel) {int fd = channel->fd;struct ChannelMap* channelMap = evLoop->channelMap;if(fd >= channelMap->size || channelMap->list[fd] == NULL) {return -1;}int ret = evLoop->dispatcher->modify(channel,evLoop);return ret;
}

总结