libuv异步任务

我就是我 2022-01-21 11:03 433阅读 0赞

libuv有一个异步机制，你可以将要执行的任务放到一个线程池里执行，从而使得主线程能够执行其他任务而不受阻塞。

将一个任务提交到线程池里的接口为：

int uv_queue_work(uv_loop_t* loop, uv_work_t* req, uv_work_cb work_cb, uv_after_work_cb after_work_cb)

接口的参数简单明了，其中值得注意的是uv\_work\_t这个数据结构，uv\_work\_t中有一个data字段，我们一般将需要用到的数据通过自定义一个struct来将数据保存到data字段中，例如下面这样：

typedef struct userdata
    {
        int userid;
        int userip;
    } userdata_t;
    
    uv_work_t* w = (uv_work_t*)malloc(sizeof(uv_work_t));
    userdata_t* ud = (userdata_t*)malloc(sizeof(userdata_t));
    
    w->data = ud;

使用时我们将真正要执行的任务放在work\_cb中执行，libuv中内置的线程池会调用我的传到uv\_queue\_work中的work\_cb函数，一但work\_cb调用完毕后，libuv中的线程池会接着调用我们传到uv\_queue\_work中的after\_work\_cb函数。

下面以一个异步调用mysql接口代码为例子，进行代码学习：

头文件：

#ifndef __MYSQL_WRAPPER_H__
    #define __MYSQL_WRAPPER_H__
    
    #include <vector>
    #include <string>
    
    class mysql_wrapper {
    public:
    	static void connect(char* ip, 
                            int port,
                            char* db_name, 
                            char* uname, 
                            char* pwd,
    						void(*open_cb)(const char* err, void* context));
    
    	static void close(void* context);
    
    	static void query(void* context,
    	                  char* sql,
    					  void(*query_cb)(const char* err,std::vector<std::vector<std::string>>* result));
    };
    
    
    #endif

实现文件：

#include <stdio.h>
    #include <string.h>
    #include <stdlib.h>
    
    #include "uv.h"
    
    #ifdef WIN32
    #include <winsock.h>
    #pragma comment(lib, "ws2_32.lib")
    #pragma comment(lib, "libmysql.lib")
    #endif
    
    #include "mysql.h"
    
    #include "mysql_wrapper.h"
    
    #define my_malloc malloc
    #define my_free free
    
    struct connect_req {
    	char* ip;
    	int port;
    
    	char* db_name;
    
    	char* uname;
    	char* upwd;
    
    	void(*open_cb)(const char* err, void* context);
    
    	char* err;
    	void* context;
    };
    
    struct mysql_context {
    	void* pConn; // mysql
    	uv_mutex_t lock; 
    
    	int is_closed;
    };
    
    static void
    connect_work(uv_work_t* req) {
    	struct connect_req* r = (struct connect_req*)req->data;
    	MYSQL* pConn = mysql_init(NULL);
    	if (mysql_real_connect(pConn, r->ip, r->uname, r->upwd, r->db_name, r->port, NULL, 0)) { // 
    		// r->context = pConn;
    		struct mysql_context* c = (struct mysql_context*)my_malloc(sizeof(struct mysql_context));
    		memset(c, 0, sizeof(struct mysql_context));
    
    		c->pConn = pConn;
    		uv_mutex_init(&c->lock);
    
    		r->context = c;
    		r->err = NULL;
    	}
    	else {
    		r->context = NULL;
    		r->err = strdup(mysql_error(pConn));
    	}
    }
    
    static void
    on_connect_complete(uv_work_t* req, int status) {
    	struct connect_req* r = (struct connect_req*)req->data;
    	r->open_cb(r->err, r->context);
    
    	if (r->ip) {
    		free(r->ip);
    	}
    	if (r->db_name) {
    		free(r->db_name);
    	}
    	if (r->uname) {
    		free(r->uname);
    	}
    	if (r->upwd) {
    		free(r->upwd);
    	}
    	if (r->err) {
    		free(r->err);
    	}
    
    	my_free(r);
    	my_free(req);
    }
    
    void 
    mysql_wrapper::connect(char* ip, int port,
    	char* db_name, char* uname, char* pwd,
    	void(*open_cb)(const char* err, void* context)) {
    	uv_work_t* w = (uv_work_t*)my_malloc(sizeof(uv_work_t));
    	memset(w, 0, sizeof(uv_work_t));
    
    	struct connect_req* r = (struct connect_req*)my_malloc(sizeof(struct connect_req));
    	memset(r, 0, sizeof(struct connect_req));
    
    	r->ip = strdup(ip);
    	r->port = port;
    	r->db_name = strdup(db_name);
    	r->uname = strdup(uname);
    	r->upwd = strdup(pwd);
    	r->open_cb = open_cb;
    
    	w->data = (void*) r;
    	uv_queue_work(uv_default_loop(), w, connect_work, on_connect_complete);
    }
    
    static void
    close_work(uv_work_t* req) {
    	struct mysql_context* r = (struct mysql_context*)(req->data);
    	uv_mutex_lock(&r->lock);
    	MYSQL* pConn = (MYSQL*)r->pConn;
    	mysql_close(pConn);
    	uv_mutex_unlock(&r->lock);
    }
    
    static void
    on_close_complete(uv_work_t* req, int status) {
    	struct mysql_context* r = (struct mysql_context*)(req->data);
    	my_free(r);
    	my_free(req);
    }
    
    void 
    mysql_wrapper::close(void* context) {
    	struct mysql_context* c = (struct mysql_context*) context;
    	if (c->is_closed) {
    		return;
    	}
    
    	uv_work_t* w = (uv_work_t*)my_malloc(sizeof(uv_work_t));
    	memset(w, 0, sizeof(uv_work_t));
    	w->data = (context);
    
    	c->is_closed = 1;
    	uv_queue_work(uv_default_loop(), w, close_work, on_close_complete);
    }
    
    struct query_req {
    	void* context;
    	char* sql;
    	void(*query_cb)(const char* err, std::vector<std::vector<std::string>>* result);
    
    	char* err;
    	std::vector<std::vector<std::string>>* result;
    };
    
    static void
    query_work(uv_work_t* req) {
    	query_req* r = (query_req*)req->data;
    
    	struct mysql_context* my_conn = (struct mysql_context*)(r->context);
    
    	uv_mutex_lock(&my_conn->lock);
    
    	MYSQL* pConn = (MYSQL*)my_conn->pConn;
    
    	
    	int ret = mysql_query(pConn, r->sql);
    	if (ret != 0) {
    		r->err = strdup(mysql_error(pConn));
    		r->result = NULL;
    		uv_mutex_unlock(&my_conn->lock);
    		return;
    	}
    
    	r->err = NULL;
    	MYSQL_RES *result = mysql_store_result(pConn);
    	if (!result) {
    		r->result = NULL;
    		return;
    	}
    
    	MYSQL_ROW row;
    
    	r->result = new std::vector<std::vector<std::string>>;
    	int num = mysql_num_fields(result);
    	std::vector<std::string> empty;
    
    	std::vector<std::vector<std::string>>::iterator end_elem;
    	while (row = mysql_fetch_row(result)) {
    		r->result->push_back(empty);
    		end_elem = r->result->end() - 1;
    		for (int i = 0; i < num; i++) {
    			end_elem->push_back(row[i]);
    		}
    	}
    
    	mysql_free_result(result);
    	uv_mutex_unlock(&my_conn->lock);
    }
    
    static void
    on_query_complete(uv_work_t* req, int status) {
    	query_req* r = (query_req*)req->data;
    	r->query_cb(r->err, r->result);
    
    	if (r->sql) {
    		free(r->sql);
    	}
    
    	if (r->result) {
    		delete r->result;
    	}
    
    	if (r->err) {
    		free(r->err);
    	}
    
    	my_free(r);
    	my_free(req);
    }
    
    void 
    mysql_wrapper::query(void* context,
                         char* sql,
    					 void(*query_cb)(const char* err, std::vector<std::vector<std::string>>* result)) {
    	struct mysql_context* c = (struct mysql_context*) context;
    	if (c->is_closed) {
    		return;
    	}
    
    	uv_work_t* w = (uv_work_t*)my_malloc(sizeof(uv_work_t));
    	memset(w, 0, sizeof(uv_work_t));
    
    	query_req* r = (query_req*)my_malloc(sizeof(query_req));
    	memset(r, 0, sizeof(query_req));
    	r->context = context;
    	r->sql = strdup(sql);
    	r->query_cb = query_cb;
    
    	w->data = r;
    	uv_queue_work(uv_default_loop(), w, query_work, on_query_complete);
    }

以头文件中的代码为线索浏览实现文件中的代码，可以发现在代码还是很简单的。mysql接口用的是mysql官方的connector/c接口，代码可以从mysql官方中下载。