libuv 高性能事件驱动库简介

青旅半醒 2021-08-29 02:59 580阅读 0赞

libuv是一个高性能事件驱动库，屏蔽了各种操作系统的差异从而提供了统一的API。libuv严格使用异步、事件驱动的编程风格。其核心工作是提供事件循环及基于 I/O或其他活动事件的回调机制。libuv库包含了诸如计时器、非阻塞网络支持、异步文件系统访问、线程创建、子进程等核心工具。  
libuv有二个主要功能，一个是循环调度模块，也就是异步IO的核心Loop模块，一个是全局的线程池Thread Pool，loop模块主要是用于异步通知，thread pool主要用于线程管理和调度。libuv库里边用到线程池的地方有DNS相关的二个函数getaddrinfo和getnameinfo，文件各种异步操作。其他暂时不使用thread pool，thread pool主要是给调度者使用。

## 如何安装、使用libuv这个框架 ##

首先我们可以在[libuv][]上找到libuv这个框架，在README.md里，我们就可以在Build Instructions找到安装方法，作者电脑操作系统是macox（所以后面的实例也是以linux、unix为主，不会讲windows）。我们首先把项目clone到我们的电脑上，在项目根目录执行一下的命令，在执行过程中可能会出现各种底层库没有安装的情况，按照提示自行安装就可以了，作者在执行 xcodebuild 的时候发现不能加上 -target All 的参数，不加的话可以顺利build过去。

$ ./gyp_uv.py -f xcode
    $ xcodebuild -ARCHS="x86_64" -project uv.xcodeproj \
         -configuration Release -target All

build完成后 我们可以在项目目录里找到 build/Release/libuv.a 文件，这个就是编译后的文件了，我们稍后会用到。 准备工作做好之后我们就可以创建一个C或者C++的工程了，在Mac上我一般使用xcode来编写oc、c、c++的项目。 首先创建一个C项目，这个时候我们需要把我们之前编译的libuv.a的文件加入到项目的依赖中，我们在Build Phases中的 Link Binary with Libraries中添加libuv.a的路径，同时我们需要在项目根目录引入uv.h等文件头。准备工作做好之后，我们就开始学习怎么写标准的hello world了 哈哈哈哈。

#include <stdio.h>
    #include <stdlib.h>
    #include <uv.h>
    
    int main() {
        uv_loop_t *loop = malloc(sizeof(uv_loop_t));
        uv_loop_init(loop);
    
        printf("Now quitting.\n");
        uv_run(loop, UV_RUN_DEFAULT);
    
        uv_loop_close(loop);
        free(loop);
        return 0;
    }

上述代码仅仅初始化了一个loop循环，并没有执行任何内容，然后就close且退出了。虽然上述代码并没有利用libuv的async功能，但是给我们展示了 uv\_loop\_init uv\_run 两个核心函数。我们稍后会介绍他们做了什么。

## 先从一个数据结构开始 ##

在开始介绍整个整个libuv之前，我不得不首先介绍一个数据结构，因为这个数据结构在libuv里无处不在，这个数据结构就是--循环双向链表。 我们在项目根目录下的src目录可以找到queue.h的头文件。不错，这个数据结构就是用宏实现的，那我让我们一起来学习一下什么是链表。

## 链表的定义： ##

链表是一种物理存储单元上非连续、非顺序的存储结构

## 那什么是双向链表呢？ ##

![format_png][]

双向链表其实就是头尾相连

## 那什么是双向循环链表呢？ ##

![format_png 1][]

看图我们就明白了，所谓的循环链表就是把头尾相连。

## 来看一下 queue.h 是怎么实现的 ##

#define QUEUE_NEXT(q)       (*(QUEUE **) &((*(q))[0]))
    #define QUEUE_PREV(q)       (*(QUEUE **) &((*(q))[1]))
    #define QUEUE_PREV_NEXT(q)  (QUEUE_NEXT(QUEUE_PREV(q)))
    #define QUEUE_NEXT_PREV(q)  (QUEUE_PREV(QUEUE_NEXT(q)))
    
    /* Public macros. */
    #define QUEUE_DATA(ptr, type, field)                                          \
      ((type *) ((char *) (ptr) - offsetof(type, field)))
    
    #define QUEUE_INIT(q)                                                         \
      do {                                                                        \
        QUEUE_NEXT(q) = (q);                                                      \
        QUEUE_PREV(q) = (q);                                                      \
      }                                                                           \
      while (0)

上述代码我只截取了部分的实现 其实这里我只想讲两个点 1：QUEUE\_NEXT 的实现

(*(QUEUE **) &((*(q))[0]))

在这个宏里，他为什么用这个复杂的方式来实现呢？ 其实他有两个目的：强制类型转换、成为左值

*(q))[0]

这个步骤是取到数组的第一个元素

(QUEUE **)

这个步骤进行强制类型转换

(*(nnn) &(xxx))

这个步骤目的就是为了使xxx成为左值

2：QUEUE\_DATA 获取链表的值 巧妙的使用了地址的偏移量来完成

## 来看一个使用queue.h的demo吧 ##

#include "queue.h"
    #include <stdio.h>
    
    static QUEUE* q;
    static QUEUE queue;
    
    struct user_s {
        int age;
        char* name;
        QUEUE node;
    };
    
    int main() {
        struct user_s* user;
        struct user_s john;
        struct user_s henry;
    
        john.name = "john";
        john.age = 44;
        henry.name = "henry";
        henry.age = 32;
    
        QUEUE_INIT(&queue);
        QUEUE_INIT(&john.node);
        QUEUE_INIT(&henry.node);
        QUEUE_INIT(&willy.node);
        QUEUE_INIT(&sgy.node);
    
        ((*(&queue))[0]) = john.node;
        (*(QUEUE **) &((*(&queue))[0])) = &john.node;
    
        QUEUE_INSERT_TAIL(&queue, &john.node);
        QUEUE_INSERT_TAIL(&queue, &henry.node);
    
        q = QUEUE_HEAD(&queue);
    
        user = QUEUE_DATA(q, struct user_s, node);
    
        printf("Received first inserted user: %s who is %d.\n",
               user->name, user->age);
    
        QUEUE_REMOVE(q);
    
        QUEUE_FOREACH(q, &queue) {
            user = QUEUE_DATA(q, struct user_s, node);
    
            printf("Received rest inserted users: %s who is %d.\n",
                   user->name, user->age);
        }
    
        return 0;
    }

从上面代码可以总结出5个方法 *QUEUE\_INIT* 队列初始化 *QUEUE\_INSERT\_TAIL* 插入到队尾 *QUEUE\_HEAD* 头部第一个元素 *QUEUE\_DATA* 获得元素的内容 *QUEUE\_REMOVE* 从队列中移除元素

那双向循环链表就先简单介绍到这。

## libuv的核心 ##

libuv为什么可以这么高效呢？实际他使用了操作系统提供的高并发异步模型

linux: epoll

freebsd: kqueue

windows: iocp

每个我们常见的操作系统都为我们封装了类似的高并发异步模型，那libuv其实就是对各个操作系统进行封装，最后暴露出统一的api供开发者调用，开发者不需要关系底层是什么操作系统，什么API了。 我们来看一下同步模型和异步模型的区别

## 阻塞模型 ##

![format_png 2][]

我们在一个线程中调用网络请求，之后线程就会被阻塞，直到返回结果才能继续执行线程

## 异步模型 ##

![format_png 3][]

在异步模型中 我们调用网络请求后不在去直接调用accept阻塞线程，而是轮询fd是否发生变化，在返回内容后我们在调用cb执行我们的代码，这个过程是非阻塞的。 说了这么多我们通过2个例子了解一下其中的原理。

## 学习如何建立一个socket ##

我们首先了解一下 C是如何创建socket的，之后我们在看一下如果通过高并发异步模型来创建socket，最后我们在了解一下 libuv下怎么创建socket。

## C如何创建一个socket呢？ ##

#include <sys/types.h>
    #include <sys/socket.h>
    #include <stdio.h>
    #include <netinet/in.h>
    #include <arpa/inet.h>
    #include <unistd.h>
    #include <string.h>
    #include <stdlib.h>
    #include <fcntl.h>
    #include <sys/shm.h>
    
    #define MYPORT  8887
    #define QUEUE   20
    #define BUFFER_SIZE 1024
    
    int main()
    {
        //定义sockfd AF_INET（IPv4） AF_INET6（IPv6） AF_LOCAL（UNIX协议） AF_ROUTE（路由套接字） AF_KEY（秘钥套接字）
        // SOCK_STREAM（字节流套接字） SOCK_DGRAM
        int server_sockfd = socket(AF_INET, SOCK_STREAM, 0);
    
        ///定义sockaddr_in
        struct sockaddr_in server_sockaddr;
        server_sockaddr.sin_family = AF_INET;
        server_sockaddr.sin_port = htons(MYPORT);
        server_sockaddr.sin_addr.s_addr = htonl(INADDR_ANY);
    
        ///bind，成功返回0，出错返回-1
        if(bind(server_sockfd,(struct sockaddr *)&server_sockaddr,sizeof(server_sockaddr))==-1)
        {
            perror("bind");
            exit(1);
        }
    
        printf("监听%d端口\n", MYPORT);
        ///listen，成功返回0，出错返回-1
        if(listen(server_sockfd, QUEUE) == -1)
        {
            perror("listen");
            exit(1);
        }
    
        ///客户端套接字
        char buffer[BUFFER_SIZE];
        struct sockaddr_in client_addr;
        socklen_t length = sizeof(client_addr);
    
        printf("等待客户端连接\n");
        ///成功返回非负描述字，出错返回-1
        int conn = accept(server_sockfd, (struct sockaddr*)&client_addr, &length);
        if(conn<0)
        {
            perror("connect");
            exit(1);
        }
        printf("客户端成功连接\n");
    
        while(1)
        {
            memset(buffer,0,sizeof(buffer));
            long len = recv(conn, buffer, sizeof(buffer), 0);
            //客户端发送exit或者异常结束时，退出
            ;
            if(strcmp(buffer,"exit\n")==0 || len<=0) {
                printf("出现异常");
                break;
            }
    
            printf("来自客户端数据：\n");
            fwrite(buffer, len, 1, stdout);
            send(conn, buffer, len, 0);
            printf("发送给客户端数据：\n");
            fwrite(buffer, len, 1, stdout);
        }
        close(conn);
        close(server_sockfd);
        return 0;
    }

代码一大坨，其实上我们简单拆分一下

第一步：创建socket 文件描述符
    第二步：定义socket addr
    第三步：绑定文件描述符和地址  bind
    第四步：监听文件描述符 listen
    第五步：等待socket返回内容 accept
    第六步：接收信息 recv

## 那我们如何使用kqueue来创建socket呢？ ##

由于作者电脑是macos，所以只能使用kqueue，不能使用epoll。

#include <sys/types.h>
    #include <sys/socket.h>
    #include <stdio.h>
    #include <netinet/in.h>
    #include <arpa/inet.h>
    #include <unistd.h>
    #include <string.h>
    #include <stdlib.h>
    #include <fcntl.h>
    #include <sys/shm.h>
    
    #define MYPORT  8887
    #define QUEUE   20
    #define BUFFER_SIZE 1024
    
    int main()
    {
        // 定义sockfd AF_INET（IPv4） AF_INET6（IPv6） AF_LOCAL（UNIX协议） AF_ROUTE（路由套接字） AF_KEY（秘钥套接字）
        // SOCK_STREAM（字节流套接字） SOCK_DGRAM
        int server_sockfd = socket(AF_INET, SOCK_STREAM, 0);
    
        // 定义sockaddr_in
        struct sockaddr_in server_sockaddr;
        server_sockaddr.sin_family = AF_INET;
        server_sockaddr.sin_port = htons(MYPORT);
        server_sockaddr.sin_addr.s_addr = htonl(INADDR_ANY);
    
        // bind，成功返回0，出错返回-1
        if(bind(server_sockfd,(struct sockaddr *)&server_sockaddr,sizeof(server_sockaddr))==-1)
        {
            perror("bind");
            exit(1);
        }
    
        printf("监听%d端口\n", MYPORT);
        // listen，成功返回0，出错返回-1
        if(listen(server_sockfd, QUEUE) == -1)
        {
            perror("listen");
            exit(1);
        }
    
        //创建一个消息队列并返回kqueue描述符
        int kq =  kqueue();
        struct kevent change_list;  //想要监控的事件
        struct kevent event_list[10000];  //用于kevent返回
        char buffer[1024];
        int nevents;
        // 监听sock的读事件
        EV_SET(&change_list, server_sockfd, EVFILT_READ, EV_ADD | EV_ENABLE, 0, 0, 0);
        while(1) {
            printf("new loop...\n");
            // 等待监听事件的发生
            nevents = kevent(kq, &change_list, 1, event_list, 2, NULL);
            if (nevents < 0) {
                printf("kevent error.\n");  // 监听出错
            } else if (nevents > 0) {
                printf("get events number: %d\n", nevents);
                for (int i = 0; i < nevents; ++i) {
                    printf("loop index: %d\n", i);
                    struct kevent event = event_list[i]; //监听事件的event数据结构
                    int clientfd = (int) event.ident;  // 监听描述符
                    // 表示该监听描述符出错
                    if (event.flags & EV_ERROR) {
                        close(clientfd);
                        printf("EV_ERROR: %s\n", strerror(event_list[i].data));
                    }
                    // 表示sock有新的连接
                    if (clientfd == server_sockfd) {
                        printf("new connection\n");
                        struct sockaddr_in client_addr;
                        socklen_t client_addr_len = sizeof(client_addr);
                        int new_fd = accept(server_sockfd, (struct sockaddr *) &client_addr, &client_addr_len);
                        long len = recv(new_fd, buffer, sizeof(buffer), 0);
                        char remote[INET_ADDRSTRLEN];
                        printf("connected with ip: %s, port: %d\n",
                               inet_ntop(AF_INET, &client_addr.sin_addr, remote, INET_ADDRSTRLEN),
                               ntohs(client_addr.sin_port));
                        send(new_fd, buffer, len, 0);
                    }
                }
            }
        }
        return 0;
    }

我们可以看到，listen之前都是一样的，不在赘述，简化一下后面的步骤

第一步：创建 kqueue描述符
    第二部：监听socket读事件 EV_SET
    第三步：绑定kq 和 change_list kevent

一直while循环直到kevent返回可以的文件描述符数量 那到这里其实我们就完全弄懂了 如何直接用C写出高并发异步是怎么运行的。那么我们就看看使用libuv的例子吧

## 使用libuv的scoket ##

#include <stdio.h>
    #include <stdlib.h>
    #include <string.h>
    #include <uv.h>
    
    #define DEFAULT_PORT 7000
    #define DEFAULT_BACKLOG 128
    
    uv_loop_t *loop;
    struct sockaddr_in addr;
    
    typedef struct {
        uv_write_t req;
        uv_buf_t buf;
    } write_req_t;
    
    void free_write_req(uv_write_t *req) {
        write_req_t *wr = (write_req_t*) req;
        free(wr->buf.base);
        free(wr);
    }
    
    void alloc_buffer(uv_handle_t *handle, size_t suggested_size, uv_buf_t *buf) {
        buf->base = (char*) malloc(suggested_size);
        buf->len = suggested_size;
    }
    
    void on_close(uv_handle_t* handle) {
        free(handle);
    }
    
    void echo_write(uv_write_t *req, int status) {
        if (status) {
            fprintf(stderr, "Write error %s\n", uv_strerror(status));
        }
        free_write_req(req);
    }
    
    void echo_read(uv_stream_t *client, ssize_t nread, const uv_buf_t *buf) {
        if (nread > 0) {
            write_req_t *req = (write_req_t*) malloc(sizeof(write_req_t));
            req->buf = uv_buf_init(buf->base, nread);
            fwrite(buf->base, 30, 1, stdout);
            uv_write((uv_write_t*) req, client, &req->buf, 1, echo_write);
            return;
        }
        if (nread < 0) {
            if (nread != UV_EOF)
                fprintf(stderr, "Read error %s\n", uv_err_name(nread));
            uv_close((uv_handle_t*) client, on_close);
        }
    
        free(buf->base);
    }
    
    void on_new_connection(uv_stream_t *server, int status) {
        if (status < 0) {
            fprintf(stderr, "New connection error %s\n", uv_strerror(status));
            // error!
            return;
        }
    
        uv_tcp_t *client = (uv_tcp_t*) malloc(sizeof(uv_tcp_t));
        uv_tcp_init(loop, client);
        if (uv_accept(server, (uv_stream_t*) client) == 0) {
            uv_read_start((uv_stream_t*) client, alloc_buffer, echo_read);
        }
        else {
            uv_close((uv_handle_t*) client, on_close);
        }
    }
    
    int main() {
        loop = uv_default_loop();
    
        uv_tcp_t server;
        uv_tcp_init(loop, &server);
    
        uv_ip4_addr("0.0.0.0", DEFAULT_PORT, &addr);
    
        uv_tcp_bind(&server, (const struct sockaddr*)&addr, 0);
        int r = uv_listen((uv_stream_t*) &server, DEFAULT_BACKLOG, on_new_connection);
        if (r) {
            fprintf(stderr, "Listen error %s\n", uv_strerror(r));
            return 1;
        }
        return uv_run(loop, UV_RUN_DEFAULT);
    }

实际上整体我们都可以把libuv和我们原生的c kqueue进行一一对应，发现相差不多，唯一不同是我们需要定义 uv\_loop 这个内部循环，后面我们在来讲套循环机制。

## 学习如何进行文件读写 ##

我们学习完了网络，那么我们再来看看文件i/o是怎么处理的。

![format_png 4][]

刚刚我们玩转了socket来看这张图是不是很熟悉？但是发现右侧有了很大的不同。文件操作、DNS、用户代码不是基于epoll这种模型吗？ 显而易见我们有了答案，这是为什么呢？其实很简单文件的很多操作就是同步的，但是libuv为了统一异步，利用开辟线程进行文件等操作模拟了异步的过程！！原来我们用了这么久才发现他是个骗子。哈哈！其实是我们学艺不精。 那其实讲到这里文件读写其实讲的差不多了，我们还是来看看例子吧！

#include <stdio.h>
    #include <uv.h>
    
    uv_fs_t open_req;
    uv_fs_t _read;
    
    static char buffer[1024];
    static uv_buf_t iov;
    
    void on_read(uv_fs_t *req) {
        printf("%s\n",iov.base);
    }
    void on_open(uv_fs_t *req) {
        printf("%zd\n",req->result);
        iov = uv_buf_init(buffer, sizeof(buffer));
        uv_fs_read(uv_default_loop(), &_read, (int)req->result,
                   &iov, 1, -1, on_read);
    }
    int main() {
        const char* path = "/Users/sgy/koa/package.json";
        // O_RDONLY 、 O_WRONLY 、 O_RDWR 、 O_CREAT
        uv_fs_open(uv_default_loop(), &open_req, path, O_RDONLY, 0, on_open);
        uv_run(uv_default_loop(), UV_RUN_DEFAULT);
        uv_fs_req_cleanup(&open_req);
        return 0;
    }

其实libuv底层对文件open和read的操作是分开的。 看到这里文件api没啥讲的了，我们来简单讲讲线程池。

## 线程池 ##

线程池就是对线程的统一管理，预先创建出线程，如果有任务就把任务放到线程池里去执行。

![format_png 5][]

通过上图我们可以看到有任务进来首先会插入到链表中进行排队等待， 直到线程空余就会去链表中去取。 通过阅读 src/threadpool.c文件我们可以了解 MAX\_THREADPOOL\_SIZE 128 最大线程为128个 default\_threads\[4\] 默认只会开辟4个线程 如果你对底层不了解 那当你在进行大量的文件i/o时 线程池数量就是阻碍你的最大障碍。 为啥最大只能创建128个线程呢？因为大多数操作系统创建一个线程大概花费1M的内存空间，外加用户本身代码也要占用大量的内存，所以这里设置了最大128的限制。

## 了解libuv的循环机制 ##

我们通过网络和文件了解了libuv，那么我们来看看libuv的循环机制

uv_loop_t *loop;
     loop = uv_default_loop()
     uv_run(loop, UV_RUN_DEFAULT);

首先我们会创建 loop 然后一系列的骚操作之后 最后我们执行了uv\_run 嗯嗯 那uv\_run 肯定是突破口了 在src/unix/core.c 文件里 我们找到了 uv\_run的定义

int uv_run(uv_loop_t* loop, uv_run_mode mode) {
      int timeout;
      int r;
      int ran_pending;
    
      r = uv__loop_alive(loop);
      if (!r)
        uv__update_time(loop);
    
      while (r != 0 && loop->stop_flag == 0) {
        uv__update_time(loop);
        uv__run_timers(loop);
        ran_pending = uv__run_pending(loop);
        uv__run_idle(loop);
        uv__run_prepare(loop);
    
        timeout = 0;
        if ((mode == UV_RUN_ONCE && !ran_pending) || mode == UV_RUN_DEFAULT)
          timeout = uv_backend_timeout(loop);
    
        uv__io_poll(loop, timeout);
        uv__run_check(loop);
        uv__run_closing_handles(loop);
    
        if (mode == UV_RUN_ONCE) {
          /* UV_RUN_ONCE implies forward progress: at least one callback must have
           * been invoked when it returns. uv__io_poll() can return without doing
           * I/O (meaning: no callbacks) when its timeout expires - which means we
           * have pending timers that satisfy the forward progress constraint.
           *
           * UV_RUN_NOWAIT makes no guarantees about progress so it's omitted from
           * the check.
           */
          uv__update_time(loop);
          uv__run_timers(loop);
        }
    
        r = uv__loop_alive(loop);
        if (mode == UV_RUN_ONCE || mode == UV_RUN_NOWAIT)
          break;
      }

从代码中 我们就可以总结出libuv的运行周期 通过while循环不断的查询 loop中是否有停止符 如果有则退出 否则就不停的进行循环。

![format_png 6][]

上面的图已经清楚的描述我们uv\_run的流程了 那其中的核心 就在\*uvio\_poll\* 中 例如在 src/unix/linux-core.c 中的uvio\_poll函数 我们就可以找到 我们 epoll 熟悉的身影了。实现逻辑也和我们之前使用过的差不多。

[libuv]: http://link.zhihu.com/?target=https%3A//github.com/libuv/libuv
[format_png]: /images/20210728/b6e52d6d1fd14bacb072dd7196ad3a99.png
[format_png 1]: /images/20210728/2e0af592bc56477bbfd123ef869ca346.png
[format_png 2]: /images/20210728/1ca5d183b402410fb5d4225402184965.png
[format_png 3]: /images/20210728/92406c3f049e403284b67b3084f3abe3.png
[format_png 4]: /images/20210728/3a247265c0874136abd72251a2b0d206.png
[format_png 5]: /images/20210728/47ba3fdeccf3458db0bae5a4596ff79a.png
[format_png 6]: /images/20210728/c843d250013f42f98551ad8085ea02c3.png