在libuv中，流操作是很多操作的基础。比如读写管道、TCP/UDP套接字。在libuv的流操作中需要对输入持续流进行监控，一旦输入流有新的内容到来，便会进行空间申请、拿取数据、数据处理等一系列的操作。下面以读写管道为例开始介绍libuv的流操作:

1.API函数介绍：

1.1.开始输入流的监控

int uv_read_start(uv_stream_t*,uv_alloc_cb alloc_cb,uv_read_cb read_cb);

参数1:要监控的对象可以是uv_pipe_t、uv_udp_t、uv_tcp_t等。

参数2:回调函数，用于申请存储读到的内容的空间;

参数3:回调函数，用于消息处理。

1.2.往输入流写内容

int uv_write(uv_write_t* req,uv_stream_t* handle,const uv_buf_t bufs[],unsigned int nbufs,uv_write_cb cb);

参数1:写文件流对象;

参数2：要写的目标对象；

参数3：已经初始化完成的指向uv_buf_t结构体的指针；

参数4：一般为1；

参数5：写数据完成后的回调函数。

1.3.初始化管道

int uv_pipe_init(uv_loop_t, uv_pipe_t handle, int ipc);

参数1：event-loop;

参数2：与管道关联的对象；

参数3：一般写为0，具体作用在后面进行介绍；

1.4.打开/绑定管道

int uv_pipe_open(uv_pipe_t*, uv_file file);

参数1：与管道关联的对象；

参数2：文件描述符。

1.5.关闭流

void uv_close(uv_handle_t* handle, uv_close_cb close_cb);

参数1：流对象，可以是uv_pipe_t、uv_udp_t、uv_tcp_t等。

参数2：回调函数，一般为NULL(uv_close、uv_fs_close等函数都可认为是同步的)

2.回调函数介绍：

2.1.uv_read_start回调函数介绍

2.1.1.alloc_cb回调函数介绍

函数指针原型：

void (uv_alloc_cb)(uv_handle_t handle,size_t suggested_size,uv_buf_t* buf);

handle：上下文传递的对象

suggested_size：libuv建议的应该申请的空间的大小

buf：uv_buf_t结构体指针，在此回调函数里要进行的任务就是申请空间，并填充这个结构体。

2.1.2.read_cb回调函数介绍

函数指针原型：

void (uv_read_cb)(uv_stream_t stream,ssize_t nread,const uv_buf_t* buf);

nread：读到的字节数

buf：对应alloc_cb回调函数里的buf，buf->base代表了存储读取到的数据的内容的首地址。

此函数存在的目的便是对接收到的数据进行数据处理(需要注意的是数据处理时间不能太长)。

2.2.uv_write回调函数介绍

函数指针原型:

void (uv_write_cb)(uv_write_t req, int status);

其中req为传递的上下文对象，

status表征是否成功写入。为0代表成功，其他值代表失败，可用uv_strerror函数打印失败原因。

3.用libuv实现tee命令的代码

#include <stdio.h>
#include <fcntl.h>
#include <unistd.h>
#include <string.h>
#include <stdlib.h>
 
#include <uv.h>
 
typedef struct {
    uv_write_t req;
    uv_buf_t buf;
} write_req_t;
 
uv_loop_t *loop;
uv_pipe_t stdin_pipe;
uv_pipe_t stdout_pipe;
uv_pipe_t file_pipe;
 
void alloc_buffer(uv_handle_t *handle, size_t suggested_size, uv_buf_t *buf) {
    *buf = uv_buf_init((char*) malloc(suggested_size), suggested_size);
}
 
void free_write_req(uv_write_t *req) {
    write_req_t *wr = (write_req_t*) req;
    free(wr->buf.base);
    free(wr);
}
 
void on_stdout_write(uv_write_t *req, int status) {
    free_write_req(req);
}
 
void on_file_write(uv_write_t *req, int status) {
    free_write_req(req);
}
 
void write_data(uv_stream_t *dest, size_t size, uv_buf_t buf, uv_write_cb cb) {
    write_req_t *req = (write_req_t*) malloc(sizeof(write_req_t));
    req->buf = uv_buf_init((char*) malloc(size), size);
    memcpy(req->buf.base, buf.base, size);
    uv_write((uv_write_t*) req, (uv_stream_t*)dest, &req->buf, 1, cb);
}
 
void read_stdin(uv_stream_t *stream, ssize_t nread, const uv_buf_t *buf) {
    if (nread < 0){
        if (nread == UV_EOF){
            // end of file
            uv_close((uv_handle_t *)&stdin_pipe, NULL);
            uv_close((uv_handle_t *)&stdout_pipe, NULL);
            uv_close((uv_handle_t *)&file_pipe, NULL);
        }
    } else if (nread > 0) {
        write_data((uv_stream_t *)&stdout_pipe, nread, *buf, on_stdout_write);
        write_data((uv_stream_t *)&file_pipe, nread, *buf, on_file_write);
    }
 
    // OK to free buffer as write_data copies it.
    if (buf->base)
        free(buf->base);
}
 
int main(int argc, char **argv) {
    loop = uv_default_loop();
 
    uv_pipe_init(loop, &stdin_pipe, 0);
    uv_pipe_open(&stdin_pipe, 0);
 
    uv_pipe_init(loop, &stdout_pipe, 0);
    uv_pipe_open(&stdout_pipe, 1);
    
    uv_fs_t file_req;
    int fd = uv_fs_open(loop, &file_req, argv[1], O_CREAT | O_RDWR, 0644, NULL);
    uv_pipe_init(loop, &file_pipe, 0);
    uv_pipe_open(&file_pipe, fd);
 
    uv_read_start((uv_stream_t*)&stdin_pipe, alloc_buffer, read_stdin);
 
    uv_run(loop, UV_RUN_DEFAULT);
    return 0;
}

代码的基本逻辑:

1.初始化管道1，并将管道与stdin关联起来。

2.初始化管道2，并将管道与stdout关联起来。

3.初始化管道3，并将管道与打开的文件关联起来。

4.开始监控stdin输入管道。

5.在stdin有数据到来时，调用alloc_buffer回调函数，在堆上申请一定的空间。并将数据存入新申请的空间。

6.调用read_stdin函数进行消息处理，在本例中的消息处理便是将stdin的内容分别写到标准输出和文件。

7.在读到文件末尾时，关闭流，整个程序运行完毕。