声明:本文为原创博文,转载请注明出处。
Nodejs编程是全异步的,这就意味着我们不必每次都阻塞等待该次操作的结果,而事件完成(就绪)时会主动回调通知我们。在网络编程中,一般都是基于Reactor线程模型的变种,无论其怎么演化,其核心组件都包含了Reactor实例(提供事件注册、注销、通知功能)、多路复用器(由操作系统提供,比如kqueue、select、epoll等)、事件处理器(负责事件的处理)以及事件源(linux中这就是描述符)这四个组件。一般,会单独启动一个线程运行Reactor实例来实现真正的异步操作。但是,依赖操作系统提供的系统调用来实现异步是有局限的,比如在Reactor模型中我们只能监听到:网络IO事件、signel(信号)、超时事件以及一些管道事件等,但这些事件也只是通知我们资源可读或者可写,真正的读写操作(read和write)还是同步的(也就是你必须等到read或者write返回,虽然linux提供了aio,但是其有诸多槽点),那么Nodejs的全异步是如何做到的呢?你可能会很快想到,就是启用单独的线程来做同步的事情,这也是libuv的设计思路,借用官网的一张图,说明一切:
由上图可以看到,libuv实现了一套自己的线程池来处理所有同步操作(从而模拟出异步的效果),下面就来看一下该线程池的具体实现吧!
一、线程池模型
说道线程池,在java领域中,jdk本身就提供了多种线程池实现,几乎所有的线程池都遵循以下模型(任务队列+线程池):
libuv自身定义了一个非常精炼、高效的队列(双向循环链表),只用了几个简单的宏定义将其实现,具体实现方式可以参见我的另一篇博文:libuv高效队列的实现。现在队列有了,来看一下task的定义:
1 struct uv__work { 2 void (*work)(struct uv__work *w); 3 void (*done)(struct uv__work *w, int status); 4 struct uv_loop_s* loop; 5 void* wq[2]; 6 };
uv__work就代表一个task,可以看到里面有两个函数指针(work代表任务实际操作,done用于对任务进行状态确认)。wq成员就是一个QUEUE的节点, uv__work就是通过wq与其他 uv__work连接成一个队列。
下面来看一下threadpool的初始化,代码如下:
1 #define MAX_THREADPOOL_SIZE 128 2 3 static uv_once_t once = UV_ONCE_INIT; 4 static uv_cond_t cond; 5 static uv_mutex_t mutex; 6 static unsigned int idle_threads;//当前空闲的线程数 7 static unsigned int nthreads; 8 static uv_thread_t* threads; 9 static uv_thread_t default_threads[4]; 10 static QUEUE exit_message; 11 static QUEUE wq;//线程池全部会检查这个queue,一旦发现有任务就执行,但是只能有一个线程抢占到 12 static volatile int initialized; 13 14 15 static void init_once(void) { 16 unsigned int i; 17 const char* val; 18 // 线程池中的线程数,默认值为4 19 nthreads = ARRAY_SIZE(default_threads); 20 val = getenv("UV_THREADPOOL_SIZE"); 21 if (val != NULL) 22 nthreads = atoi(val); 23 if (nthreads == 0) 24 nthreads = 1; 25 if (nthreads > MAX_THREADPOOL_SIZE) 26 nthreads = MAX_THREADPOOL_SIZE; 27 28 threads = default_threads; 29 if (nthreads > ARRAY_SIZE(default_threads)) { 30 // 分配线程句柄 31 threads = uv__malloc(nthreads * sizeof(threads[0])); 32 if (threads == NULL) { 33 nthreads = ARRAY_SIZE(default_threads); 34 threads = default_threads; 35 } 36 } 37 // 初始化条件变量 38 if (uv_cond_init(&cond)) 39 abort(); 40 // 初始化互斥锁 41 if (uv_mutex_init(&mutex)) 42 abort(); 43 44 // 初始化任务队列 45 QUEUE_INIT(&wq); 46 47 // 创建nthreads个线程 48 for (i = 0; i < nthreads; i++) 49 if (uv_thread_create(threads + i, worker, NULL)) 50 abort(); 51 52 initialized = 1; 53 }
上面的代码中,一共创建了nthreads个线程,那么每个线程的执行代码是什么呢?由线程创建代码:uv_thread_create(threads + i, worker, NULL),可以看到,每一个线程都是执行worker函数,下面看看worker函数都在做什么:
1 /* To avoid deadlock with uv_cancel() it‘s crucial that the worker 2 * never holds the global mutex and the loop-local mutex at the same time. 3 */ 4 static void worker(void* arg) { 5 struct uv__work* w; 6 QUEUE* q; 7 8 (void) arg; 9 10 for (;;) { 11 // 因为是多线程访问,因此需要加锁同步 12 uv_mutex_lock(&mutex); 13 14 // 如果任务队列是空的 15 while (QUEUE_EMPTY(&wq)) { 16 // 空闲线程数加1 17 idle_threads += 1; 18 // 等待条件变量 19 uv_cond_wait(&cond, &mutex); 20 // 被唤醒之后,说明有任务被post到队列,因此空闲线程数需要减1 21 idle_threads -= 1; 22 } 23 24 // 取出队列的头部节点(第一个task) 25 q = QUEUE_HEAD(&wq); 26 27 if (q == &exit_message) 28 uv_cond_signal(&cond); 29 else { 30 // 从队列中移除这个task 31 QUEUE_REMOVE(q); 32 QUEUE_INIT(q); /* Signal uv_cancel() that the work req is 33 executing. */ 34 } 35 36 uv_mutex_unlock(&mutex); 37 38 if (q == &exit_message) 39 break; 40 41 // 取出uv__work首地址 42 w = QUEUE_DATA(q, struct uv__work, wq); 43 // 调用task的work,执行任务 44 w->work(w); 45 46 uv_mutex_lock(&w->loop->wq_mutex); 47 w->work = NULL; /* Signal uv_cancel() that the work req is done 48 executing. */ 49 QUEUE_INSERT_TAIL(&w->loop->wq, &w->wq); 50 uv_async_send(&w->loop->wq_async); 51 uv_mutex_unlock(&w->loop->wq_mutex); 52 } 53 }
可以看到,多个线程都会在worker方法中等待在conn条件变量上,一旦有任务加入队列,线程就会被唤醒,然后只有一个线程会得到任务的执行权,其他的线程只能继续等待。
那么如何向队列提交一个task呢?看以下代码:
1 void uv__work_submit(uv_loop_t* loop, 2 struct uv__work* w, 3 void (*work)(struct uv__work* w), 4 void (*done)(struct uv__work* w, int status)) { 5 uv_once(&once, init_once); 6 // 构造一个task 7 w->loop = loop; 8 w->work = work; 9 w->done = done; 10 // 将其插入任务队列 11 post(&w->wq); 12 }
接着看post做了什么:
1 static void post(QUEUE* q) { 2 // 同步队列操作 3 uv_mutex_lock(&mutex); 4 // 将task插入队列尾部 5 QUEUE_INSERT_TAIL(&wq, q); 6 // 如果当前有空闲线程,就向条件变量发送信号 7 if (idle_threads > 0) 8 uv_cond_signal(&cond); 9 uv_mutex_unlock(&mutex); 10 }
有提交任务,就肯定会有取消一个任务的操作,是的,他就是uv__work_cancel,代码如下:
1 static int uv__work_cancel(uv_loop_t* loop, uv_req_t* req, struct uv__work* w) { 2 int cancelled; 3 4 uv_mutex_lock(&mutex); 5 uv_mutex_lock(&w->loop->wq_mutex); 6 7 // 只有当前队列不为空并且要取消的uv__work有效时才会继续执行 8 cancelled = !QUEUE_EMPTY(&w->wq) && w->work != NULL; 9 if (cancelled) 10 QUEUE_REMOVE(&w->wq);// 从队列中移除task 11 12 uv_mutex_unlock(&w->loop->wq_mutex); 13 uv_mutex_unlock(&mutex); 14 15 if (!cancelled) 16 return UV_EBUSY; 17 18 // 更新这个task的状态 19 w->work = uv__cancelled; 20 uv_mutex_lock(&loop->wq_mutex); 21 QUEUE_INSERT_TAIL(&loop->wq, &w->wq); 22 uv_async_send(&loop->wq_async); 23 uv_mutex_unlock(&loop->wq_mutex); 24 25 return 0; 26 }
至此,一个线程池的组成以及实现原理都说完了,可以看到,libuv几乎是用了最少的代码完成了高效的线程池,这对于我们平时写代码时具有很好的借鉴意义,文中涉及到uv_req_t以及uv_loop_t等结构我都直接跳过,因为这牵扯到libuv的其他组件,我将在以后的源码剖析中逐步阐述,谢谢你能看到这里。