半同步/半反应堆模型（使用线程池）的TCP服务器例子

红太狼 2022-08-05 02:59 240阅读 0赞

在半同步/半异步模式中“同步”和“异步”与I/O模型中同步、异步的概念不同：I/O模型中，同步和异步区分的是内核向应用程序通知的是何种I/O事件（是就绪事件还是完成事件），以及该由谁来完成I/O读写（是应用程序还是内核）。在并发模式中，“同步”指的是程序完全按照代码序列的顺序执行；“异步”指的是程序的执行需要由系统事件来驱动（常见的系统事件包括中断、信号）。

异步线程的执行的执行效率高，实时性强，但是编写以异步方式执行的程序相对复杂，难以调试和扩展，而且不适合大量的并发。而同步线程虽然效率相对较低，但是逻辑简单。对于服务器这种既要求较好的实时性，又要求能同时处理多个客户请求的应用程序，可以同时使用同步线程和异步线程实现。**异步线程**监听客户请求，将其封装成请求对象插入请求队列中。请求队列通知某个在**同步模式下的工作线程**来读取并处理该对象。最简单的选工作线程的是Round Robin算法，也可以使用条件变量或信号量。  
如图是半同步/半异步的变形：“半同步/半反应堆模型”：  
![半同步/半反应堆模型][20150604102241525]

上图所示模型缺点：（1）主线程和工作线程共享请求队列。主线程添加任务和工作线程从队列取任务都要对请求队列加锁，消耗ＣＰＵ时间。（２）每个工作线程在同一时间仅处理一个客户请求。如果客户数据多，而工作线程少，则请求队列中任务堆积，客户响应会越来越慢。可以通过增加工作线程来解决。
    下图是相对高效的半同步半异步模式：

![高效方式][20150604102837689]

该模式的主线程只管理监听ｓｏｃｋｅｔ，连接ｓｏｃｋｅｔ由工作线程来管理。当有新的连接到来时，主线程就接受并将新返回的连接ｓｏｃｋｅｔ派发给某个工作线程，此后该新ｓｏｃｋｅｔ上的任何Ｉ／Ｏ操作都由被选中的线程来处理，直到客户关闭连接。上图的每个线程都维护自己的事件循环，它们各自独立监听不同事件。因此每个线程都是异步模式。  
　　  
　　 下面代码所示的服务器程序使用主线程来监听套接字并接收数据，将接收的数据存放在一个队列中（通过获取互斥锁来在队列中存放元素），在队列中存入数据后发送信号到条件变量信号，然后释放互斥锁，以允许线程池中某个线程为这个客户服务。线程池里的工作线程通过试图获取互斥锁，获取队列元素并对其进行处理（这里仅仅在控制台显示数据）。

//http_parse.cpp
    #include "http_parse.h"
    
    //错误处理函数
    void err_exit(const char *info) {
        perror(info);
        exit(-1);
    }
    
    //设置描述符为非阻塞模式
    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;
    }
    
    //向epoll中添加要监听的描述符
    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 );
    }
    
    //任务队列，主线程收到的数据队列，等待线程池线程处理
    queue<char *> taskqueue; 
    //用于为任务队列加锁的互斥锁和通知的条件信号量
    pthread_mutex_t clifd_mutex = PTHREAD_MUTEX_INITIALIZER;
    pthread_cond_t clifd_cond = PTHREAD_COND_INITIALIZER;
    vector<pthread_t> thread_pool(3); //线程池
    
    
    int main(int argc, char **argv) {
        struct sockaddr_in servaddr;
        memset(&servaddr, 0, sizeof(servaddr));
        servaddr.sin_family = AF_INET;
        servaddr.sin_addr.s_addr = htonl(INADDR_ANY);
        servaddr.sin_port = htons(9877);
    
        int listenfd = socket(AF_INET, SOCK_STREAM, 0);
        if (-1 == listenfd)
            err_exit("socket error");
        int opt = 1;
        if (-1 == setsockopt(listenfd, SOL_SOCKET, SO_REUSEADDR, &opt, sizeof(opt)))
            err_exit("setsockopt error");
        if (-1 == bind(listenfd, (const struct sockaddr *)&servaddr,
            sizeof(servaddr)))
            err_exit("bind error");
        if (-1 == setnonblocking(listenfd))
            err_exit("make_socket_non_blocking error");
        if (-1 == listen(listenfd, 1000))
            err_exit("listen error");
    
        struct epoll_event event, events[MAX_EVENTS];
        int epollfd = epoll_create(5);
        if (-1 == epollfd)
            err_exit("epoll_create error");
        event.data.fd = listenfd;
        event.events = EPOLLIN;
        if (-1 == epoll_ctl(epollfd, EPOLL_CTL_ADD, listenfd, &event))
            err_exit("epoll_ctl error");
    
        //创建线程池
        int i;
        for (i = 0; i < 3; ++i)
            thread_make(i);
    
        for ( ; ; ) {
            int i;
            int nfds = epoll_wait(epollfd, events, MAX_EVENTS, -1);
            for (i = 0; i < nfds; ++i) {
                if ((events[i].events & EPOLLERR) || (events[i].events & EPOLLHUP) ||
                    !(events[i].events & EPOLLIN)) {
                        fprintf(stderr, "epoll error\n");
                        continue;
                }
                else if (listenfd == events[i].data.fd) {
                    for ( ; ; ) {
                        struct sockaddr cliaddr;
                        socklen_t clilen = sizeof(cliaddr);
                        int connfd = accept(listenfd, (struct sockaddr*)&cliaddr, &clilen);
                        if (-1 == connfd) {
                            if ((errno == EAGAIN) || (errno == EWOULDBLOCK))
                                break;
                            else
                                err_exit("accept error");
                        }
                        if (-1 == setnonblocking(connfd))
                            err_exit("setnonblocking connfd error");
                        event.data.fd = connfd;
                        event.events = EPOLLIN | EPOLLET; //边缘触发、读事件
                        if (-1 == epoll_ctl(epollfd, EPOLL_CTL_ADD, connfd, &event))
                            err_exit("epoll_ctl add connfd error");
                    }
                    continue;
                }
                else if(events[i].events & EPOLLIN ) {
                    int done = 0;
                    int nread = 0;
                    char buf[65535];
                    int count;
                    while((count = read(events[i].data.fd, buf + nread, sizeof(buf))) > 0) {
                        nread += count;
                    }
                    if (count == -1 && errno != EAGAIN) {
                        done = 1;
                        err_exit("read error");
                    }
                    else if (0 == count)
                        done = 1;
                    buf[nread] = '\0';
    
    #ifdef DEBUG
                    write(STDOUT_FILENO, buf, nread);
                    write(STDOUT_FILENO, "\n", 1);
                    printf("nread: %d\n", nread);
                    write(STDOUT_FILENO, "\n\n", 2);
    #endif
    
                    pthread_mutex_lock(&clifd_mutex);
                    if (strlen(buf) > 4)
                        taskqueue.push(buf);
                    //taskqueue.push(string(buf));
                    pthread_cond_signal(&clifd_cond);
                    pthread_mutex_unlock(&clifd_mutex);
    
    
                    if (done) {
    #ifdef DEBUG
                        printf("close: %d\n", events[i].data.fd);
    #endif
                        epoll_ctl(epollfd, EPOLL_CTL_DEL, events[i].data.fd, 0);
                    }
                }
                else {}
            }
        }
        close(listenfd);
        return 0;
    }
    
    //线程创建函数
    void thread_make(int i) {
        void *thread_worker(void *);
        pthread_create(&thread_pool[i], NULL, &thread_worker, (void*)i);
        return;
    }
    
    //回调函数，供线程创建函数调用
    void *thread_worker(void *arg) {
    #ifdef DEBUG
        printf("thread %d starting\n", (int)arg);
    #endif
    
        for ( ; ; ) {
            pthread_mutex_lock(&clifd_mutex);
            //如果任务队列为空则等待队列数据到来
            while (0 == taskqueue.size())
                pthread_cond_wait(&clifd_cond, &clifd_mutex);
    #ifdef DEBUG
            printf("\ndd %s\n\n", taskqueue.front());
    #endif
            //将队列首元素出队
            taskqueue.pop();
            pthread_mutex_unlock(&clifd_mutex);
        }
    }

//http_parse.h
    #ifndef HTTP_PARSE_H
    #define HTTP_PARSE_H
    
    #include <queue>
    #include <sys/types.h>
    #include <sys/socket.h>
    #include <netinet/in.h>
    #include <arpa/inet.h>
    #include <assert.h>
    #include <stdio.h>
    #include <unistd.h>
    #include <errno.h>
    #include <string.h>
    #include <fcntl.h>
    #include <stdlib.h>
    #include <sys/epoll.h>
    #include <signal.h>
    #include <sys/wait.h>
    #include <sys/mman.h>
    #include <sys/stat.h>
    #include <fcntl.h>
    #include <string>
    #include <pthread.h>
    #define MAXNCLI 100
    #define MAX_EVENTS 1024
    #define SERV_PORT 9877
    using namespace std;
    
    void thread_make(int i);
    int iput, iget;
    int setnonblocking(int fd);
    void addfd(int epollfd, int fd);
    void err_exit(const char *info);
    #endif

测试服务器的客户端是文章[《unix网络编程》（15）poll函数以及使用poll的客户服务器程序][unix_15_poll_poll]中的客户端。  
　　下图是服务端测试截图：  
![服务器测试][20150604103929340]

[20150604102241525]: /images/20220805/b212dc689c1b41ac8ee78dde771c7d6c.png
[20150604102837689]: /images/20220805/0cee1a8272cc47de8ac81b4147065372.png
[unix_15_poll_poll]: http://blog.csdn.net/u013074465/article/details/44901969
[20150604103929340]: /images/20220805/df5a055109754bb3b18fd6c9c16e2889.png