curl多线程大批量分片下载大文件源码示例

本文主要是介绍curl多线程大批量分片下载大文件源码示例，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！

这段时间，一直在探索使用curl多线程来下载一系列的大文件的可行性方法。下面是我探索的结果：

1.将大文件分为许多小片段，比如20M一个片段（当然这个值可以配置，比如100M一个片段，取决于你的业务需求），使用http range来下载这些片段；

2.使用预先生成的线程池来连续不间断地执行文件片段的下载，这个线程池也就是一个固定线程个数的普通线程池，使用互斥锁和信号量来保护任务队列，各worker线程抢占式获取要下载的任务；

3.文件的下载使用磁盘文件形式，当然你也可以修改为保存在内存中。

4.由于easy handle不能跨线程调用，我们只能是一个线程函数里一个curl easy handle，用完就释放掉。

5.如果是小文件（2M以内的，比如html，img之类的），建议使用单线程中的curl multi handle的异步调用架构，后续我会将相关代码放到网上。

由于一个文件分为多个分片，每个分片可能被不同的线程下载，每个分片完成的次序并不固定，这和Bittorrent有些相似，所以，文件的关闭是个需要审慎考虑的问题。怎样处理，才能保证所有的文件全都及时安全关闭呢？及时的要求是，整个文件一下载，最好就把它关闭，以免占用文件描述符。

在目前demo的设计中，我还没有找到一个很好的思路来将它们及时关闭。欢迎大牛指点良策。

源码示例的功能：

目前的代码中，设置了是否启用http range下载的宏开关，同时定义了每个分片大小是20M。在开启range分片下载时，一个63M的文件，将会分为 63 /20 = 3 + 1 =4个分片。因而在线程池中可能会被4个线程下载。而目前线程池我设置了30个线程。

如果没有开启range分片下载，就是一个线程下载一个文件，这个文件有可能是2M，也有可能是400M，还有可能是2.8G，所以会造成大多数线程空闲，个别线程累死，这对合理高效利用计算资源是不利的，所以，我提倡尽可能使用range下载大文件。

下面是源码

//g++ -g test_pool.cpp thread_pool.cpp -o test_pool -lpthread -lcurl
//
#include <unistd.h>
#include <libgen.h>
#include <curl/curl.h>
#include <string>
#include <vector>
#include <map>
#include "thread_pool.h"using namespace std;#define  HTTP_RANGE    0 //是否开启http range下载功能?
const long seg_size = 20 * 1024 * 1024; //文件分片阈值大小是100MB, 超过该大小的文件将会强制分片下载static pthread_mutex_t g_mutex = PTHREAD_MUTEX_INITIALIZER;
map<string, int>  download_map; //下载任务的完成情况, key是url或是文件名, val是分片数, 用于主线程监视class CloseInfo {public:CloseInfo(FILE* f, string& s): fp(f), url(s) {}~CloseInfo(){}public:FILE*    fp;string   url;
};
vector<CloseInfo*> closed_vec; //所有打开的文件列表需要最后继中关闭, 因为在分片下载的情况下,无法保证在所有的分片都下载完成的情况下关闭文件//这里必须是struct而不能是class
struct JobInfo{JobInfo(): fp(NULL), startPos(0), stopPos(0), ranged(false) {}FILE*    fp;  //本地文件句柄long     startPos;long     stopPos;string   url;bool     ranged; //是否需要http range下载?
};static void set_JobInfo(JobInfo* ji, FILE* f, long s, long e, string u, bool r){ji->fp = f;ji->startPos = s;ji->stopPos = e;ji->url = u;ji->ranged = r;
}long get_download_file_length (const char *url)
{double file_len = 0;CURL *handle = curl_easy_init ();curl_easy_setopt (handle, CURLOPT_URL, url);curl_easy_setopt (handle, CURLOPT_FOLLOWLOCATION, 1L);curl_easy_setopt (handle, CURLOPT_MAXREDIRS, 3L);curl_easy_setopt (handle, CURLOPT_AUTOREFERER, 1L);curl_easy_setopt (handle, CURLOPT_HEADER, 0L);   //只需要header头curl_easy_setopt (handle, CURLOPT_NOBODY, 1L);   //不需要bodycurl_easy_setopt (handle, CURLOPT_FORBID_REUSE, 1);curl_easy_setopt (handle, CURLOPT_USERAGENT, "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/36.0.1985.143 Safari/537.36"); //user-agentif (curl_easy_perform (handle) == CURLE_OK){curl_easy_getinfo (handle, CURLINFO_CONTENT_LENGTH_DOWNLOAD, &file_len);}else{file_len = -1;}curl_easy_cleanup(handle);return file_len;
}//每个线程在下载每个文件分片时,都会回调该函数
static size_t write_data(void* ptr, size_t size, size_t nmemb, void* userdata){JobInfo* ji = (JobInfo*) userdata;bool ranged = ji->ranged;size_t written;//要分片下载的大文件, 需要设置http range域if(ranged){//多线程写同一个文件, 需要加锁pthread_mutex_lock (&g_mutex);if(ji->startPos + size * nmemb <= ji->stopPos){fseek(ji->fp, ji->startPos, SEEK_SET);written = fwrite(ptr, size, nmemb, ji->fp);ji->startPos += size * nmemb;}else{fseek(ji->fp, ji->startPos, SEEK_SET);written = fwrite(ptr, 1, ji->stopPos - ji->startPos + 1, ji->fp);ji->startPos = ji->stopPos;}pthread_mutex_unlock (&g_mutex);}else{written = fwrite(ptr, size, nmemb, ji->fp);}return written;
}void* job_process(void* arg){CURL* curl;CURLcode res;JobInfo* ji = (JobInfo*)arg;char range[64] = { 0 };if(ji->ranged) {snprintf (range, sizeof(range), "%ld-%ld", ji->startPos, ji->stopPos);printf("range: [%s]\n", range);}curl = curl_easy_init();curl_easy_setopt(curl, CURLOPT_URL, ji->url.c_str());curl_easy_setopt(curl, CURLOPT_WRITEFUNCTION, write_data);curl_easy_setopt(curl, CURLOPT_WRITEDATA, (void*)ji);curl_easy_setopt(curl, CURLOPT_FOLLOWLOCATION, 1L);curl_easy_setopt(curl, CURLOPT_MAXREDIRS, 3L);curl_easy_setopt(curl, CURLOPT_AUTOREFERER, 1L);curl_easy_setopt(curl, CURLOPT_NOSIGNAL, 1L);curl_easy_setopt(curl, CURLOPT_LOW_SPEED_LIMIT, 1L);curl_easy_setopt(curl, CURLOPT_LOW_SPEED_TIME, 5L);curl_easy_setopt(curl, CURLOPT_USERAGENT, "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/36.0.1985.143 Safari/537.36"); //user-agentcurl_easy_setopt(curl, CURLOPT_FORBID_REUSE, 1L);//curl_easy_setopt(curl, CURLOPT_NOPROGRESS, 0L);if(ji->ranged)curl_easy_setopt(curl, CURLOPT_RANGE, range);res = curl_easy_perform(curl);if(CURLE_OK != res){printf("[%d] %s\n", res, ji->url.c_str());}curl_easy_cleanup(curl);//TODO: 回馈下载分片的完成情况....string url = ji->url;pthread_mutex_lock(&g_mutex);download_map[url]--;printf("[-1]%s\n", url.c_str());pthread_mutex_unlock(&g_mutex);return NULL;
}//从url列表得到一个相应的下载任务列表
int get_job_queue (vector<string>& urls, vector<JobInfo*>& jobs){FILE* fp;long file_len = 0L, start, stop, seg_num;bool ranged = false;JobInfo* ji;vector<string>::iterator it;for(it = urls.begin(); it != urls.end(); ++it){string url = *it;file_len = get_download_file_length (url.c_str());if(file_len <= 0) continue;printf("[%ld] %s\n", file_len, url.c_str());//对应每个url, 打开一个本地文件const char* fn = basename((char*)url.c_str());string full_path(fn);full_path = "./" + full_path; //设为当前路径下面fp = fopen(full_path.c_str(), "wb"); //在此统一打开文件if(NULL == fp)  continue;//构造关闭文件向量CloseInfo* ci = new CloseInfo(fp, url);closed_vec.push_back(ci);//加入全局任务监听映射int additional = (file_len % seg_size == 0) ? 0 : 1;int seg_total = ranged ? (file_len < seg_size ? 1 : (file_len/seg_size + additional)) : 1;download_map[url] = seg_total;printf("[+1]%s, seg total %d\n", url.c_str(), seg_total);#if HTTP_RANGE//根据文件大小, 确定是否分片?if(file_len < seg_size){start = 0;stop = file_len - 1;ji = new JobInfo();set_JobInfo(ji, fp, start, stop, url, ranged);jobs.push_back(ji);}else{//分片下载,先确定分片个数ranged = true;seg_num = (long)file_len / seg_size;printf("filesize[%ld], segsize[%ld], seg num: %ld\n", file_len, seg_size, seg_num);for(int i = 0; i <= seg_num; i++){if(i < seg_num){start = i * seg_size;stop = (i + 1) * seg_size - 1;}else{if(file_len % seg_size != 0){start = i * seg_size;stop = file_len - 1;}elsebreak;}ji = new JobInfo();set_JobInfo(ji, fp, start, stop, url, ranged);jobs.push_back(ji);}}
#elsestart = 0;stop = file_len - 1;ji = new JobInfo();set_JobInfo(ji, fp, start, stop, url, ranged);jobs.push_back(ji);
#endif}return 0;
}int main(int argc, char* argv[]){vector<string> urls_vec;urls_vec.push_back("http://dlsw.baidu.com/sw-search-sp/soft/da/17519/BaiduYunGuanjia_4.8.3.1409021519.exe");//10M// urls_vec.push_back("http://cdimage.ubuntu.com/releases/14.04/release/ubuntu-14.04-server-ppc64el.template"); //67M// urls_vec.push_back("http://cdimage.ubuntu.com/releases/14.04/release/ubuntu-14.04.1-server-amd64+mac.template"); //88Murls_vec.push_back("http://cdimage.ubuntu.com/releases/14.04/release/ubuntu-14.04.1-server-powerpc.template"); //64M// urls_vec.push_back("http://cdimage.ubuntu.com/releases/14.04/release/ubuntu-14.04.1-server-ppc64el.template"); //67M// urls_vec.push_back("http://cdimage.ubuntu.com/releases/14.04/release/ubuntu-14.04-desktop-amd64+mac.iso"); //962Murls_vec.push_back("http://cdimage.ubuntu.com/releases/14.04/release/ubuntu-14.04-desktop-amd64+mac.iso.zsync"); //1.9M// urls_vec.push_back("http://cdimage.ubuntu.com/releases/14.04/release/ubuntu-14.04-server-amd64+mac.iso"); //566M// urls_vec.push_back("http://cdimage.ubuntu.com/releases/14.04/release/ubuntu-14.04-server-amd64+mac.template");  //88M// urls_vec.push_back("http://cdimage.ubuntu.com/releases/14.04/release/ubuntu-14.04-server-powerpc.iso"); //654M//创建线程越多, cpu占用率越大, 下载效率会有一定提高CThreadPool* pool = new CThreadPool(30);vector<JobInfo*> ji_vec;get_job_queue(urls_vec, ji_vec);vector<JobInfo*>::iterator it;for(it = ji_vec.begin(); it != ji_vec.end(); ++it){JobInfo* ji = *it;pool->pool_add_job(job_process, (void*)ji);}//这个时间仅是一个粗略值,当有线程下载文件未完成时,这个worker线程是不会退出的, master线程会一直等待, 直到所有的worker线程都退出//此处更好的思路是统计所有的任务是否下载完成, 再准备退出//usleep(2000 * 1000 * 1000); //2000map<string, int>::iterator mit;int finished = 0;size_t map_size = download_map.size();printf("map size %ld\n", map_size);while(1){finished = 0;pthread_mutex_lock(&g_mutex);for(mit = download_map.begin(); mit != download_map.end(); ++mit)if(!mit->second)  finished++;pthread_mutex_unlock(&g_mutex);if(finished == map_size) {printf("finish all task =====================>\n");break;}else{usleep(30 * 1000 * 1000);}}download_map.clear();delete pool;urls_vec.clear();//注意所含的jobinfo已经在线程处理完后就删除了,这里不用再单独删除.JobInfo对象析构的时候,会关闭打开的文件.//另外线程池销毁时,会将剩下没有处理的任务删除ji_vec.clear();//打开的文件,要记得全部关闭vector<CloseInfo*>::iterator cit;for(cit = closed_vec.begin(); cit != closed_vec.end(); ++cit){CloseInfo* ci = *cit;if(ci->fp != NULL){fclose(ci->fp);ci->fp = NULL;}printf("[closed] %s\n", ci->url.c_str());delete ci;}closed_vec.clear();return 0;
}

注意需依赖libcurl和lpthread库, 要先安装好libcurl库.另外,固定线程池的源码参见我另一篇博客,这里仅是替换了test_pool.cpp的代码, 改为curl下载文件.

简单C++线程池包装类源码示例

运行截图