注册    登录
关注
Py学习  »  NGINX

动手打造Nginx多进程架构

半栈工程师 • 5 年前 • 257 次点击  

最近对Nginx源码比较感兴趣，借助于强大的VS Code，我一步一步，似魔鬼的步伐，开始了Nginx的探索之旅。关于 VS Code 如何调试 Nginx 可参考上篇文章《VS CODE 轻松调试 Nginx》。

一. 引言

Nginx 其实无需做太多介绍，作为业界知名的高性能服务器，被广大互联网公司应用，阿里的 Tegine 就是基于 Nginx 开发的。

Nginx 基本上都是用来做负载均衡、反向代理和动静分离。目前大部分公司都采用 Nginx 作为负载均衡器。作为 LBS，最基本的要求就是要支持高并发，毕竟所有的请求都要经过它来进行转发。

那么为什么 Nginx 拥有如此强大的并发能力呢？这便是我感兴趣的事情，也是这篇文章所要讲的事情。但是标题是《动手打造Nginx多进程架构》，难道这篇文章却只是简单的源码分析？

这几天研究 Nginx 过程中，我常常陷于Nginx 复杂的源码之中，不得其解，虽然也翻了一些资料和书籍，但是总觉得没有 get 到精髓，就是好像已经理解了，但是对于具体流程和细节，总是模模糊糊。于是趁着周末，花了小半天，再次梳理了下Nginx 多进程事件的源码，仿照着写了一个普通的 Server，虽然代码和功能都非常简单，不过刚好适合于读者了解Nginx，而不至于陷于丛林之中，不知方向。

二. 传统 Web Server 架构

让我们来思考下，如果让你动手打造一个 web 服务器，你会怎么做？

第一步，监听端口

第二步，处理请求

监听端口倒是很简单，处理请求该怎么做呢？不知道大家上大学刚开始学c语言的时候，老师有没有布置过聊天室之类的作业？那时候我其实完全靠百度来完成的：开启端口监听，死循环接收请求，每接收一个请求就直接开个新线程去处理。

这样做当然可以，也很简单，完全满足了我当时的作业要求，其实目前很多web服务器，诸如tomcat之类，也都是这样做的，为每个请求单独分配一个线程。那么这样做，有什么弊端呢？

最直接的弊端就是线程数量开的太多，会导致 CPU 在不同线程之间不断的进行上下文切换。CPU 的每次任务切换，都需要为上一次任务保存一些上下文信息（如寄存器的值），再装载新任务的上下文信息，这些都是不小的开销。

第二个弊端就是CPU利用率的下降，考虑当前只有一个线程的情况，当线程在等待网络 IO 的时候其实是处于阻塞状态，这个时候 CPU 便处于空闲状态，这直接导致了 CPU 没有被充分利用，简直是暴殄天物！

这种架构，使 Web 服务器从骨子里，就对高并发没有很好的承载能力！

三. Nginx 多进程架构

Nginx 之所以可以支持高并发，正是因为它摒弃了传统 Web 服务器的多线程架构，并充分利用了 CPU。

Nginx采用的是 单Master、多Worker 架构，顾名思义，Master 是老板，而 Worker 才是真正干活的工人阶层。

我们先来看下 Nginx 接收请求的大概架构。

乍一看，好像和传统的 Web Server 也没啥区别啊，不过是右边的 Thread 变成了 Worker 罢了。这其实正是 Nginx 的精妙之处。

Master 进程启动后，会 fork 出 N 个 Worker 进程，N 是 可配置的，一般来说，可以设置为服务器核心数，设置更大值也没有太多意义，无非是会增加 CPU 进程切换的开销。

每个Worker 进程都会监听来自客户端的请求，并进行处理，与传统 Web Server 不同的是，Worker 进程不会对于每个请求都分配一个单独线程去处理，而是充分利用了异步 IO 的特性。

如果读者之前没有了解或者使用过异步IO，那确实该好好充充电了。Android 中的 Looper、Java 著名的开源库 Netty，都是基于异步IO，所谓异步IO，与同步IO最大的区别就是，进程不会在等待 IO 操作时被阻塞，而是可以去干其他的任务，当 IO 操作 Ready 时，操作系统会主动通知进程。

Nginx 正是使用了这样的思想，虽然同时有很多请求需要处理，但是没必要为每个请求都分配一个线程啊。哪个请求的网络 IO Ready 了，我就去处理哪个，这样不就可以了吗？何必创建一个线程在那傻傻的等着。

举个不恰当的例子，服务器就好比是学校，客户端好比是学生，学生有不会的问题就会问老师。

  • 对于传统的 Web 服务器，每个学生，学校都会派一个老师去服务，一个学校可能有几千个学生，那岂不是要雇几千个老师，校领导怕是连工资都发不出来了吧。仔细想想，每个学生不可能随时都在提问吧，总得休息下吧！那学生休息时，老师干嘛呢？白拿工资还不干活。
  • 对于Nginx，它就不给老师闲的机会啦，学校有几间办公室，就雇几个老师，有学生提问时，就派一个老师解答，所以一个老师会负责很多学生，哪个学生举手了，他就去帮助哪个学生解决问题。

这里有读者怕是会疑惑，如果哪个学生一直霸占着老师不放怎么办？这样老师不就没有机会去解答其他同学的问题了吗？如果作为一个负责业务处理的 Web 服务器，Nginx这种架构确实可能出现这样的问题，但是要记住，Nginx主要是用来做负载均衡的，他的主要任务是接收请求、转发请求，所以它的业务处理其实就是将请求再转发给其他的服务器，那么接收用异步IO，转发也用异步 IO 不就行了。

四. 源码分析

基于最新 1.15.5 版本

4.1 整体运行机制

一切都从 main（）开始。

nginx 的 main（）方法中有不少逻辑，不过对于今天我要讲的事情来说，最重要的就是两件事：

  1. 创建套接字，监听端口；
  2. Fork 出 N 个 Worker 进程。

监听端口没什么太多逻辑，我们先来看看 Worker 进程的诞生：

static void
ngx_start_worker_processes(ngx_cycle_t *cycle, ngx_int_t n, ngx_int_t type)
{
    ngx_int_t      i;
    ngx_channel_t  ch;

    ....
    for (i = 0; i < n; i++) {

        ngx_spawn_process(cycle, ngx_worker_process_cycle,
                          (void *) (intptr_t) i, "worker process", type);
        ......
    }
}

这里主要是根据配置的 Worker 数量，创建出对应数量的 Worker 进程，创建 Woker 进程调用的是 ngx_spawn_process（），第二个参数 ngx_worker_process_cycle 就是子进程的新起点。

static void
ngx_worker_process_cycle(ngx_cycle_t *cycle, void *data)
{
    ......

    for ( ;; ) {

        ......

        ngx_process_events_and_timers(cycle);

        ......
    }
}

上面的代码省略了一些逻辑，只保留了最核心的部分。ngx_worker_process_cycle ，正如其名，在其内部开启了一个死循环，不断调用 ngx_process_events_and_timers（）。

void
ngx_process_events_and_timers(ngx_cycle_t *cycle)
{
    ......

    if (ngx_use_accept_mutex) {
        if (ngx_accept_disabled > 0) {
            ngx_accept_disabled--;

        } else {
            if (ngx_trylock_accept_mutex(cycle) == NGX_ERROR) {
                return;
            }

            ......
        }
    }

    ......

    (void) ngx_process_events(cycle, timer, flags);

    ......
}

这里最后调用了ngx_process_events（）来接收并处理事件。

ngx_process_events（）在不同平台指向不同的异步 IO 处理模块，比如Linux上为epoll，而在Mac OS上指向的其实是kqueue模块中的ngx_kqueue_process_events（）。

static ngx_int_t
ngx_kqueue_process_events(ngx_cycle_t *cycle, ngx_msec_t timer,
    ngx_uint_t flags)
{
    int               events, n;
    ngx_int_t         i, instance;
    ngx_uint_t        level;
    ngx_err_t         err;
    ngx_event_t      *ev;
    ngx_queue_t      *queue;
    struct timespec   ts, *tp;

    n = (int) nchanges;
    nchanges = 0;

    ......

    events = kevent(ngx_kqueue, change_list, n, event_list, (int) nevents, tp);

    ......

    for (i = 0; i < events; i++) {

        ......

        ev = (ngx_event_t *) event_list[i].udata;

        switch (event_list[i].filter) {

        case EVFILT_READ:
        case EVFILT_WRITE:

            ......

            break;

        case EVFILT_VNODE:
            ev->kq_vnode = 1;

            break;

        case EVFILT_AIO:
            ev->complete = 1;
            ev->ready = 1;

            break;
        ......

        }
        ......

        ev->handler(ev);
    }

    return NGX_OK;
}

上面其实就是一个比较基本的 kqueue 使用方式了。说到这里，我们就不得不说下 kqueue 的使用方式了。

kqueue 主要依托于两个 API：

// 创建一个内核消息队列，返回队列描述符
int  kqueue(void); 

// 用途：注册\反注册 监听事件，等待事件通知
// kq，上面创建的消息队列描述符
// changelist，需要注册的事件
// changelist，changelist数组大小
// eventlist，内核会把返回的事件放在该数组中
// nevents，eventlist数组大小
// timeout，等待内核返回事件的超时事件，NULL 即为无限等待
int  kevent(int kq, 
	       const struct kevent *changelist, int nchanges,
	       struct kevent *eventlist, int nevents,
	       const struct timespec *timeout);

我们回过头再来看看上面 ngx_kqueue_process_events（）中代码，其实也就是在调用kevent（）等待内核返回消息，收到消息后再进行处理。这里消息处理主要是进行ACCEPT、READ、WRITE等。

所以从整体来看，Nginx事件模块的运行就是 Worker 进程在死循环中，不断等待内核消息队列返回事件消息，并加以处理的一个过程。

4.2 惊群问题

到这里我们一直在讨论一个单独的 Worker 进程运行机制，那么每个 Worker 进程之间有没有什么交互呢？

回到上面的 ngx_process_events_and_timers（）中，在每次调用 ngx_process_events（）等待消息之前，Worker 进程都会进行一个 ngx_trylock_accept_mutex（）操作，这其实就是多个 Worker 进程之间在争夺监听资格的过程，是 Nginx 为了解决惊群问题而设计出的方案。

所谓惊群，其实就是如果有多个Worker进程同时在监听内核消息事件，当有请求到来时，每个Worker进程都会被唤醒，去accept同一个请求，但是只能有一个进程会accept成功，其他进程会accept失败，被白白的唤醒了，就像你再睡觉时被突然叫醒，却发现压根没你啥事，你说气不气人。

为了解决这个问题，Nginx 让每个Worker 进程在监听内核消息事件前去竞争一把锁，只有成功获得锁的进程才能去监听内核事件，其他进程就乖乖的睡眠在锁的等待队列上。当获得锁的进程处理完accept事件，就会回来释放掉这把锁，这时所有进程又会同时去竞争锁了。

为了不让每次都是同一个进程抢到锁，Nginx 设计了一个小算法，用一个因子ngx_accept_disabled 去 平均每个进程获得锁的概率，感兴趣的同学可以自己看下这块源码。

五. 动手打造 Nginx 多进程架构

终于到DIY的环节了，这里我基于 MacOS 平台来开发，异步IO库也是选用上面所讲的 kqueue。

5.1 创建进程锁，用于抢到监听事件资格

    mm = (mt*)mmap(NULL,sizeof(*mm),PROT_READ|PROT_WRITE,MAP_SHARED|MAP_ANON,-1,0);
    memset(mm,0x00,sizeof(*mm));
    
    pthread_mutexattr_init(&mm->mutexattr);
    pthread_mutexattr_setpshared(&mm->mutexattr, PTHREAD_PROCESS_SHARED);
    pthread_mutex_init(&mm->mutex,&mm->mutexattr);

5.2 创建套接字，监听端口

   // 创建套接字
    int serverSock =socket(AF_INET, SOCK_STREAM, 0);
    if (serverSock == -1)
    {
        
        printf("socket failed\n");
        exit(0);
    }
    
    //绑定ip和端口
    struct sockaddr_in server_addr;
    server_addr.sin_family = AF_INET;
    server_addr.sin_port = htons(9999);
    server_addr.sin_addr.s_addr = inet_addr("127.0.0.1");
    if(::bind(serverSock, (struct sockaddr*)&server_addr, sizeof(server_addr)) == -1)
    {
        printf("bind failed\n");
        exit(0);
    }
    
    //启动监听
    if(listen(serverSock, 20) == -1)
    {
        printf("listen failed\n");
        exit(0);
    }

5.3 创建多个 Worker 进程

    // fork 出 3 个 Worker 进程
    int result;
    for(int i = 1; i< 3; i++){
        result = fork();
        if(result == 0){
            startWorker(i,serverSock);
            printf("start worker %d\n",i);
            break;
        }
    }

5.4 启动Worker 进程，异步监听 IO 事件




    
void startWorker(int workerId,int serverSock)
{ 
    // 创建内核事件队列
    int kqueuefd=kqueue();
    struct kevent change_list[1];  //想要监控的事件的数组
    struct kevent event_list[1];  //用来接受事件的数组

    //初始化所需注册事件
    EV_SET(&change_list[0], serverSock, EVFILT_READ, EV_ADD | EV_ENABLE, 0, 0, 0);
    
    // 循环接受事件
    while (true) {
        // 竞争锁，获取监听资格
        pthread_mutex_lock(&mm->mutex);
        printf("Worker %d get the lock\n",workerId);
        // 注册事件，等待通知
        int nevents = kevent(kqueuefd, change_list, 1, event_list, 1, NULL);
        // 释放锁
        pthread_mutex_unlock(&mm->mutex);
        //遍历返回的所有就绪事件
        for(int i = 0; i< nevents;i++){
            struct kevent event =event_list[i];
            if(event.ident == serverSock){
                // ACCEPT 事件
                handleNewConnection(kqueuefd,serverSock);
            }else if(event.filter == EVFILT_READ){
                //读取客户端传来的数据
                char * msg = handleReadFromClient(workerId,event);
                handleWriteToClient(workerId,event,msg);
            }
        }
    }
}

5.5 开启多个 Client 进程测试

void startClientId(int clientId)
{
    //创建套接字
    int sock = socket(AF_INET, SOCK_STREAM, 0);
    //向Server发起请求
    struct sockaddr_in serv_addr;
    serv_addr.sin_family = AF_INET;  //使用IPv4地址
    serv_addr.sin_addr.s_addr = inet_addr("127.0.0.1");  //具体的IP地址
    serv_addr.sin_port = htons(9999);  //端口
    connect(sock, (struct sockaddr*)&serv_addr, sizeof(serv_addr));
    
    while (true) {
        //向服务器传送数据
        string s = "I am Client ";
        s.append(to_string(clientId));
    
        char str[60];
        strcpy(str,s.c_str());
        write(sock, str, strlen(str));
        
        //读取服务器传回的数据
        char buffer[60];
        if(read(sock, buffer, sizeof(buffer)-1)>0){
            printf("Client %d receive : %s\n",clientId,buffer);
        }
        
        sleep(9);
    }
}

运行结果：

哈哈，基本实现了我的要求。

Demo 源码见：

HalfStackDeveloper/LearnNginx

六. 总结

Nginx 之所以有强大的高并发能力，得益于它与众不同的架构设计，无论是多进程还是异步IO，都是 Nginx 不可或缺的一部分。研究 Nginx 源码十分有趣，但是看源码和动手写又是两回事，看源码只能大概了解脉络，只有自己操刀，才能真正理解和运用！


今天看啥 - 高品质阅读平台
本文地址：http://www.jintiankansha.me/t/Rc2isHo81B
Python社区是高质量的Python/Django开发社区
本文地址：http://www.python88.com/topic/25630
Reply
 
257 次点击  
登录后回复
关于   移动版   ·   三行代码   ·   今天看啥   ·   Code   ·   link之家   ·   卧龙搜索   ·   藏经阁   ·   小百科
Py学习 - 专注于Python技术发展的社区(原Django社区)
沪ICP备11025650号