最近在工作中，遇到了这么一个需求：
我们希望拥有一个高性能的 http 请求处理客户端程序，这个客户端要求有这样的架构：
它拥有两个线程
一个线程接收业务程序通过消息队列发来的批量的 http 请求信息，进行批量的 http 请求
另一个线程接收外部的 http 应答，并将应答信息放到本地的消息队列中供业务程序使用
要求在 http 请求的处理过程中，尽量保持不阻塞高性能处理
这个需求的框架图大概是这个样子：
n 个业务程序
http client
n 个 http 服务端程序
消息队列：发送批量的 http 请求报文
libcurl multi: 发起批量的 http 请求
libcurl multi: 处理批量的 http 应答
消息队列：转发 http 服务端的应答报文
n 个业务程序
http client
n 个 http 服务端程序
可以看到，我们的需求，其实就是满足平台中大量的业务程序的 http 请求处理需求。其中业务程序自然不止一个，业务程序要请求的 http 服务端也不止一个，我们要做的 http 请求客户端程序实际上就像一个网关中间层。
那么现在问题来了，如果做到 http client 程序的高性能处理。我们当然不想它请求一次就阻塞掉，这样会非常非常的慢；我们也不想让处理应答这种耗时的操作占用太多时间，这样也会非常非常的慢。
此时，我想到了当初学习 libcurl 的时候在官网上读到的那句话：
The multi interface allows a single-threaded application to perform the same kinds of multiple, simultaneous transfers that multi-threaded programs can perform. It allows many of the benefits of multi-threaded transfers without the complexity of managing and synchronizing many threads.
尤其是最关键的执行请求的那个函数 curl_multi_perform 函数是异步的：
curl_multi_perform is asynchronous. It will only perform what can be done now and then return back control to your program. It is designed to never block. You need to keep calling the function until all transfers are completed.
这个接口非常适合用于我们现在的需求场景里面，当 http client 发起批量的 http 请求的时候，使用 libcurl multi 接口就可以实现不阻塞式异步处理。
那么如何设计和实现呢？
这里，我通过参考 libcurl 的官方示例 10-at-a-time.c 的代码框架，设计实现了两个设计方案（思路略有不同）的实现；并且因为工作环境的 libcurl 是老版本的库（没有 curl_multi_wait 函数的版本）而同时实现了新版本（使用 curl_multi_wait）的代码和旧版本（没有 curl_multi_wait 函数的版本）的代码。
接下来让我们开始吧：）
ps:1. 10-at-a-time.c 官方示例代码网址 https://curl.haxx.se/libcurl/c/10-at-a-time.html2. 本篇博客中所有的实现代码托管在 GitHub 上，仓库地址 https://github.com/wangying2016/libcurl_multi_http_client
在设计实现我们自己的 http client 程序之前，我们先要来看看这个让我们启发很大的官方示例 demo 10-at-a-time.c 到底实现了什么。
根据 libcurl 的版本不同，10-at-a-time.c 的代码也稍有不同，这里我只分析最新的使用 curl_multi_wait 函数的版本（这一版本的代码也相当简洁），思路都是一样的，只是实现方式不一样（如果你跟我一样，也要考虑兼容老版本的 libcurl，那么建议你去官网上下载一个 curl-7.20.0.tar，进去里面的 docs/examples/10-at-a-time.c 中去看看老版本的实现方式）。
这里大家可以通过网址 https://curl.haxx.se/libcurl/c/10-at-a-time.html 点击进去阅读 10-at-a-time.c 的代码。我这里简单分析下这个程序的设计。
通过编译：
$ gcc -o 10-at-a-time 10-at-a-time.c -lcurl
你就可以拿到 10-at-a-time 的执行文件：
$ ./10-at-a-time
你就可以发现，这个程序实现了大量的 http 请求功能。具体并发量是多少呢？10，这也是它的名字的来源。
那么它是怎么实现的呢？
这里我也不浪费口舌，直接上流程图（细节比如说 transfers 是否小于当前全部 url 总数、获取消息是否处理成功等等判断未体现在流程图中，可自行仔细研读代码）：
外层循环
Created with Raphaël 2.2.0
开始
添加 10 个初始 url 请求
curl_multi_perform 批量发起 http 请求
内层循环处理应答并且新增请求
curl_multi_wait 轮询全部请求 handle
是否还有活动请求或者还有未处理请求
结束
yes
no
内层循环
Created with Raphaël 2.2.0
内层循环开始
curl_multi_info_read 读取应答消息
curl_easy_getinfo 获取具体信息
curl_multi_remove_handle 删除请求 handle
curl_multi_add_handle 添加新请求
是否还有可处理应答信息
内层循环结束
yes
no
总的来说
就这样，10-at-a-time.c 完成了一个高并发 http 请求，并且还能动态新增请求处理的功能。
相当于就是，既可以同时烧 10 把柴火，还能把烧完了的柴火清理掉再自动添加柴火！
这不就是我想要的功能吗：）
10-at-a-time.c 是一个线程的程序，如何让其变成两个线程的框架呢，我稍稍动了脑子，设计出来了以下的方案：
线程一负责 http 请求的批量发送和接收，然后把接收到的报文存储到全局变量中去；线程二负责将线程一存储到全局变量中的应答报文进行实际的处理
既然用到了全局变量存储应答报文，那么两个线程中的同步一定要做好（mutex）。另外，这里我使用了非常简单的方式来记录是否处理：
定义一个应答报文结构，其中定义一个变量 consumable，它为 1 即为收到应答可以处理，为 2 则意味着处理完毕
首先，请求定义在全局变量中：
char *urls[] = {
"https://www.microsoft.com",
"https://opensource.org",
...
};
为了记录应答信息，我定义了一个应答报文的结构，这里没有存储应答信息，因为实在是太大了，就简单记录下是否可以供线程二处理的状态（也就是收到了应答，记录 consumable 为 1 状态，线程二看到了 consumable 为 1 就会去处理，处理完毕置为 2）：
// response struct
struct res_info {
int index;
// 0: initial 1: can consume 2: consume end
int consumable;
char url[MAX_URL_LENGTH];
};
其中 index 记录着该 url 在全局 urls 中的序号，consumable 记录着这个应答报文是否可以处理，0 是初始化状态，1 是线程一添加到了全局变量中去可供线程二处理状态，2 则是线程二处理完毕的状态。
// response
struct res_info res_list[NUM_URLS];
res_list 则是全局的应答报文存储变量，NUM_URLS 则是全部的 url 的个数（这只是一个 demo，后续可以简单的将其扩展为链表或者队列进行动态扩展）。在线程二中，我只需要遍历 res_list，读取其中 consumable 为 1 的应答报文进行处理，处理完毕后设置为 2 即标记它处理完毕。
添加请求的函数
// add request
static void add_transfer(CURLM *cm, int i)
{
CURL *eh = curl_easy_init();
curl_easy_setopt(eh, CURLOPT_WRITEFUNCTION, write_cb);
curl_easy_setopt(eh, CURLOPT_URL, urls[i]);
curl_easy_setopt(eh, CURLOPT_TIMEOUT, 10L);
curl_easy_setopt(eh, CURLOPT_PRIVATE, urls[i]);
curl_multi_add_handle(cm, eh);
}
记录请求序号的函数
// find index
int find_index_of_urls(char *url) {
int i;
for (i = 0; i < NUM_URLS; ++i) {
if (strcmp(urls[i], url) == 0)
return i;
}
return 0;
}
添加应答报文的函数
// add response
void add_response(struct res_info res) {
int i = 0;
pthread_mutex_lock(&res_mutex);
res_list[res.index] = res;
res_list[res.index].consumable = 1;
// sleep(1);
pthread_mutex_unlock(&res_mutex);
}
处理应答报文的函数
// consume response
void consume_response() {
int i;
pthread_mutex_lock(&res_mutex);
for (i = 0; i < NUM_URLS; ++i) {
if (res_list[i].consumable == 1) {
printf("Log: t2, consume response, index [%d], url [%s]\n", \
res_list[i].index, res_list[i].url);
res_list[i].consumable = 2;
}
}
pthread_mutex_unlock(&res_mutex);
}
检查是否全部处理完毕的函数
/ check consume finished
int check_consume_finished() {
int i, finished;
finished = 1;
pthread_mutex_lock(&res_mutex);
for (i = 0; i < NUM_URLS; ++i) {
if (res_list[i].consumable != 2) {
finished = 0;
break;
}
}
pthread_mutex_unlock(&res_mutex);
return finished;
}
线程一
// t1's thread function
void *fun1() {
printf("Log: t1 begin...\n");
CURLM *cm;
CURLMsg *msg;
int msgs_left = -1;
int still_alive = 1;
curl_global_init(CURL_GLOBAL_ALL);
cm = curl_multi_init();
/* Limit the amount of simultaneous connections curl should allow: */
curl_multi_setopt(cm, CURLMOPT_MAXCONNECTS, (long)MAX_PARALLEL);
for(transfers = 0; transfers < MAX_PARALLEL; transfers++)
add_transfer(cm, transfers);
do {
curl_multi_perform(cm, &still_alive);
while((msg = curl_multi_info_read(cm, &msgs_left))) {
if(msg->msg == CURLMSG_DONE) {
struct res_info res;
char *url;
CURL *e = msg->easy_handle;
curl_easy_getinfo(msg->easy_handle, CURLINFO_PRIVATE, &url);
strcpy(res.url, url);
res.index = find_index_of_urls(res.url);
add_response(res);
printf("Log: t1, add response, error number [%d], error messsage [%s], url [%s], "
"index [%d]\n", msg->data.result, curl_easy_strerror(msg->data.result), res.url, \
res.index);
curl_multi_remove_handle(cm, e);
curl_easy_cleanup(e);
}
else {
printf("Log: t1, request error, error number [%d]\n", msg->msg);
}
if(transfers < NUM_URLS)
add_transfer(cm, transfers++);
}
if(still_alive)
curl_multi_wait(cm, NULL, 0, 1000, NULL);
} while(still_alive || (transfers < NUM_URLS));
curl_multi_cleanup(cm);
curl_global_cleanup();
printf("Log: t1 end...\n");
}
线程二
// t2's thread function
void *fun2() {
int i = 0;
printf("Log: t2 begin...\n");
do {
consume_response();
sleep(1);
} while (!check_consume_finished());
printf("Log: t2 end...\n");
}
至此，我们就实现了一个简单的高性能 http 请求处理客户端 demo 程序。详细代码可以参考我的 GitHub 上的 http_client_v1.c，其中 http_client_v1_old.c 是兼容没有 curl_multi_wait 函数的 libcurl 库版本代码。
第一个方案，说实话有些不优雅，通过一个全局缓存的应答报文列表（并且通过 mutex 加锁同步）实现两个线程之间的分工。实际上，还是线程一做完了全部的工作，既负责发，又要负责收。
那么，libcurl multi 到底是否支持将 http 的请求跟处理分离到两个线程里面去呢：
ibcurl is thread safe but has no internal thread synchronization. You may have to provide your own locking should you meet any of the thread safety exceptions below.
Handles. You must never share the same handle in multiple threads. You can pass the handles around among threads, but you must never use a single handle from more than one thread at any given time.
请看加粗的地方，libcurl 官方文档中明确说明了，libcurl 不建议在两个线程中共享 handle。这就麻烦了，libcurl multi 的 handle 究竟可不可以在两个线程中共享使用呢？
我的设想是：一个线程使用 libcurl multi handle 就负责发请求，然后增加未处理请求；另一个线程使用 libcurl multi handle 就负责处理请求，然后删已处理请求。按道理说，线程一就一直在添加请求，也就是往 libcurl multi 中的 easy handle 栈中压栈，却并没有对已经存在的 easy handle 进行什么操作，料想应该不会影响线程二中对应答报文的处理吧。
所以，说不定可以？！
所以，不如我们尝试下？
T_T，说干就干！
这一方案中去掉了方案一的用来存储应答报文的全局变量，因为用不到了（线程二直接进行处理）。
应答报文结构体
// response struct
struct res_info {
int index;
// 0: initial 1: consume end
int consumable;
char url[MAX_URL_LENGTH];
};
libcurl multi handle
// multi handle
CURLM *cm;
...
int main() {
// Initialize multi handle
curl_global_init(CURL_GLOBAL_ALL);
cm = curl_multi_init();
...
// Clean mutli handle
curl_multi_cleanup(cm);
curl_global_cleanup();
...
}
当前正在处理请求请求计数
// on-going request
unsigned int doing_cnt = 0;
// cnt mutex
pthread_mutex_t cnt_mutex = PTHREAD_MUTEX_INITIALIZER;
线程一
// t1's thread function
void *fun1() {
printf("Log: t1 begin...\n");
int still_alive = 1;
/* Limit the amount of simultaneous connections curl should allow: */
curl_multi_setopt(cm, CURLMOPT_MAXCONNECTS, (long)MAX_PARALLEL);
for(transfers = 0; transfers < MAX_PARALLEL; transfers++) {
add_transfer(transfers);
printf("Log: t1, add request, index = [%d], url = [%s]\n", \
find_index_of_urls(urls[transfers]), urls[transfers]);
}
do {
curl_multi_perform(cm, &still_alive);
if(still_alive)
curl_multi_wait(cm, NULL, 0, 1000, NULL);
if(transfers < NUM_URLS && doing_cnt < MAX_PARALLEL) {
add_transfer(transfers);
printf("Log: t1, add request, index = [%d], url = [%s]\n", \
find_index_of_urls(urls[transfers]), urls[transfers]);
transfers++;
}
// sleep(1);
} while(still_alive || (transfers < NUM_URLS));
printf("Log: t1 end...\n");
}
线程二
// t2's thread function
void *fun2() {
int i = 0;
int msgs_left = -1;
CURLMsg *msg;
printf("Log: t2 begin...\n");
do {
while((msg = curl_multi_info_read(cm, &msgs_left))) {
if(msg->msg == CURLMSG_DONE) {
char *url;
CURL *e = msg->easy_handle;
curl_easy_getinfo(msg->easy_handle, CURLINFO_PRIVATE, &url);
struct res_info res;
strcpy(res.url, url);
res.index = find_index_of_urls(res.url);
res.consumable = 1;
res_list[res.index] = res;
printf("Log: t2, accept response, index [%d], url [%s], error number [%d], error messsage [%s]\n", \
res.index, res.url, ms g->data.result, curl_easy_strerror(msg->data.result));
curl_multi_remove_handle(cm, e);
curl_easy_cleanup(e);
}
else {
printf("Log: t2, request error, error number [%d]\n", msg->msg);
}
pthread_mutex_lock(&cnt_mutex);
doing_cnt--;
pthread_mutex_unlock(&cnt_mutex);
}
// sleep(1);
} while (!check_consume_finished());
printf("Log: t2 end...\n");
}
虽然我这边修改了下代码，实现了方案二的代码。但是我还是很疑惑：
1. 我并没有对 libcurl multi handle 进行同步加锁机制，但是看上去不管怎么测试，代码都运行正常。难道官网上说的，不能在多个线程中共享 libcurl 的 handle 是有一定的例外情况的？
2. 我冒 libcurl 官网之大不韪，在两个线程中共享了 libcurl multi 接口的 handle，实际上并未出现问题。难道是因为我线程一只有添加请求的操作，从而没有影响到线程二的 handle 使用的原因吗？
3. 我方案二的代码是否是有问题的代码呢？至少目前看来，在执行上没有问题，并且再多 url 请求也没有出现问题。
这些问题，目前的我解答不了，或许需要自己日后深入 libcurl 的源码中去探索答案。这里也希望有相关经验的网友能够提供给我答案，这里悉心请教了：）
方案二的代码在 GitHub 上的 http_client_v2.c 中，同样的， http_client_v2_old.c 是兼容没有 curl_multi_wait 函数的 libcurl 库版本的代码。
根据官网说明：
libcurl is thread safe but has no internal thread synchronization.
libcurl 是没有内部的同步返回措施的，因此这个方案的代码高并发一定会有问题。一个线程对 multi handle 操作增加，一个线程对 multi handle 操作删除，高并发下会有线程安全问题。
因此，方案二的代码仅供探索试错，有问题的地方也在此强调，还请注意 ：）
通过学习 libcurl multi 接口的使用示例 10-at-a-time.c 的代码，我实现了一个使用了两个线程的高性能 http 客户端请求程序：
1. 一个线程玩命发2. 一个线程吐血收
在这个过程中对 libcurl multi 接口的运行机制有了更加深入的了解。也在探索过程中实现了两个版本的代码，并且兼容了没有 curl_multi_wait 的函数的 libcurl 老版本的库代码。
短暂的探索有了一些成果，不过长期的探索仍然还需要进行。在探索过程中，总会发现自己的知识储存太薄弱，或许在我学习了更多网络相关的知识后，再回来看这些内容，会有不一样的理解吧 ^_^
To be Stronger：）
