008-C 语言中使用 libcurl 实现多线程下载

事情是这样的，笔者在实现一个下载功能的时候遇到一个问题：服务器对下载速率进行了限制，对于业务上的文件下载，是不确定大小的，只能从服务器请求下载的时候才能知道需要下载的文件大小。如果使用单线程下载的话，用户体验非常差，因此需要实现多线程下载的功能。

获取文件大小

#include <stdlib.h>
#include <curl/curl.h>
#include <unistd.h>

static size_t get_file_size(char *url) {
  size_t filesize = 0;
  CURL *curl_handle;
  curl_global_init(CURL_GLOBAL_ALL);
  curl_handle = curl_easy_init();
  if (curl_handle) {
    curl_easy_setopt(curl_handle, CURLOPT_URL, url);
    curl_easy_setopt(curl_handle, CURLOPT_HEADER,
                     1);  // 设置 HEADER 会得到 header，文件大小信息就在 header 中
    curl_easy_setopt(curl_handle, CURLOPT_NOBODY,
                     1);  // 设置 NOBODY 可以避免下载
    CURLcode res_code = curl_easy_perform(curl_handle);  // 请求
    curl_easy_getinfo(curl_handle, CURLINFO_CONTENT_LENGTH_DOWNLOAD_T,
                      &filesize);  // 从请求头中获取待下载文件的大小
  }
  curl_global_cleanup();
  
  return filesize;
}

int main() {
  size_t file_size = 0, per_thread_size = 0;
  char url[] = "https://test.com/test.exe";

  file_size = get_file_size(url);
  printf("file_size=%d\n",file_size);

  return 0;
}

使用 gcc 编译的时候需要链接上 curl 库

gcc curl.c -o curltest -lcurl

输出结果

pureos@pureos:~/Documents$ ./curltest 
HTTP/1.1 200 OK
Server: nginx/1.21.5
Date: Mon, 08 Jan 2024 06:21:35 GMT
Content-Type: application/octet-stream
Content-Length: 74194069
Connection: keep-alive
Accept-Ranges: bytes
Content-Security-Policy: block-all-mixed-content
ETag: "2bf56162523c85f740adebd078eb3a78"
Last-Modified: Tue, 02 Jan 2024 03:02:13 GMT
Strict-Transport-Security: max-age=31536000; includeSubDomains
Vary: Origin
Vary: Accept-Encoding
X-Amz-Request-Id: 17A84AF9CDE1356D
X-Content-Type-Options: nosniff
X-Xss-Protection: 1; mode=block

file_size=74194069

检查了一下，获取到的文件大小是正确的，那么就可以直接开始多线程下载了。

多线程下载

这里我使用的是创建多个线程，每个线程下载一部分文件，全部下载完成后再合并为一个文件。

static void create_thread(void *(*routine)(void *), void *arg,
                          const pthread_attr_t *attr) {
  pthread_t tid;
  if (!pthread_create(&tid, attr, routine, arg)) {
    pthread_detach(tid);
  } else {
    printf("create thread fail!\n");
    exit(1);
  }
  usleep(300000);
  return;
}

完整代码

#include <pthread.h>
#include <stdlib.h>
#include <curl/curl.h>
#include <unistd.h>

#define THREAD_COUNT 16

typedef struct ThreadData {
  char *url;
  char *outputFilename;
  long startRange;
  long endRange;
  char finish;
} ThreadData;

/**
 * @description: 获取时间戳函数
 * @return {*}
 */
static long long get_timestamp(void) {
  long long tmp;
  struct timeval tv;

  gettimeofday(&tv, NULL);
  tmp = tv.tv_sec;
  tmp = tmp * 1000;
  tmp = tmp + (tv.tv_usec / 1000);

  return tmp;
}

/**
 * @description: 创建线程
 * @param {void} *
 * @return {*}
 */
static void create_thread(void *(*routine)(void *), void *arg,
                          const pthread_attr_t *attr) {
  pthread_t tid;
  if (!pthread_create(&tid, attr, routine, arg)) {
    pthread_detach(tid);
  } else {
    printf("create thread fail!\n");
    exit(1);
  }
  usleep(300000);
  return;
}

static size_t write_data(void *ptr, size_t size, size_t nmemb, void *stream) {
  size_t written = fwrite(ptr, size, nmemb, (FILE *)stream);
  return written;
}

/**
 * @description: 下载文件的线程
 * @param {void} *ptr
 * @return {*}
 */
static void *download_part(void *ptr) {
  ThreadData *data = (ThreadData *)ptr;
  char range[64];
  CURL *curl_handle;
  FILE *pagefile;
  CURLcode res = -1;

  snprintf(range, sizeof(range), "%ld-%ld", data->startRange, data->endRange);

  curl_global_init(CURL_GLOBAL_ALL);
  curl_handle = curl_easy_init();
  if (curl_handle) {
    curl_easy_setopt(curl_handle, CURLOPT_URL, data->url);

    // curl_easy_setopt(curl_handle, CURLOPT_VERBOSE, 1L);
    /* 这个选项用于开启详细模式（verbose mode）。设置为非零值时，libcurl
     * 会打印额外的调试信息，这些信息包括发送和接收的数据，如 HTTP 请求和响应头。
     *这对于开发和调试非常有用，因为你可以看到发生在传输层的所有事情。*/

    // curl_easy_setopt(curl_handle, CURLOPT_NOPROGRESS, 1L);
    /* 但是如果你启用了 CURLOPT_NOPROGRESS 选项并设置为零值，libcurl
     * 将调用一个进度回调函数（progress function）来显示传输进度。*/
    curl_easy_setopt(curl_handle, CURLOPT_WRITEFUNCTION, write_data);
    curl_easy_setopt(curl_handle, CURLOPT_TIMEOUT, 600L);
    curl_easy_setopt(curl_handle, CURLOPT_FOLLOWLOCATION, 1L);
    curl_easy_setopt(curl_handle, CURLOPT_RANGE, range);

    /* open the file */
    pagefile = fopen(data->outputFilename, "wb");
    if (pagefile) {
      curl_easy_setopt(curl_handle, CURLOPT_WRITEDATA, pagefile);
      res = curl_easy_perform(curl_handle);
      fclose(pagefile);
    }
    curl_easy_cleanup(curl_handle);
  }
  curl_global_cleanup();

  if (res == CURLE_OK) {
    data->finish = 1;
  }
  return NULL;
}

/**
 * @description: 合并文件
 * @param {char} *final_output
 * @param {int} num_files
 * @return {*}
 */
static int merge_files(const char *final_output, int num_files) {
  FILE *fout, *ftmp;
  char buffer[1024];
  size_t bytes;

  // 打开最终输出文件
  fout = fopen(final_output, "wb");
  if (!fout) {
    perror("Error opening final output file");
    return -1;
  }

  // 循环遍历所有临时文件
  for (int i = 0; i < num_files; ++i) {
    char tmp_file[16] = {0};
    snprintf(tmp_file, sizeof(tmp_file), "test%d.exe", i);
    // 打开临时文件
    ftmp = fopen(tmp_file, "rb");
    if (!ftmp) {
      perror("Error opening temporary file");
      fclose(fout);
      return -1;
    }

    // 读取临时文件内容并写入到最终文件中
    while ((bytes = fread(buffer, 1, sizeof(buffer), ftmp)) > 0) {
      fwrite(buffer, 1, bytes, fout);
    }

    // 关闭临时文件
    fclose(ftmp);

    // 删除临时文件
    remove(tmp_file);
  }

  // 关闭最终文件
  fclose(fout);

  return 0;
}

/**
 * @description: 获取待下载文件大小
 * @param {char} *url
 * @return {*}
 */
static size_t get_file_size(char *url) {
  size_t filesize = 0;
  CURL *curl_handle;
  curl_global_init(CURL_GLOBAL_ALL);
  curl_handle = curl_easy_init();
  if (curl_handle) {
    curl_easy_setopt(curl_handle, CURLOPT_URL, url);
    curl_easy_setopt(curl_handle, CURLOPT_HEADER,
                     1);  // 设置 HEADER 会得到 header，文件大小信息就在 header 中
    curl_easy_setopt(curl_handle, CURLOPT_NOBODY,
                     1);  // 设置 NOBODY 可以避免下载
    CURLcode res_code = curl_easy_perform(curl_handle);  // 请求
    curl_easy_getinfo(curl_handle, CURLINFO_CONTENT_LENGTH_DOWNLOAD_T,
                      &filesize);  // 从请求头中获取待下载文件的大小
    printf("file_size=%d\n", filesize);
  }
  curl_global_cleanup();

  return filesize;
}

int main() {
  ThreadData threadData[THREAD_COUNT];
  const char *final_output = "test.exe";
  long long curr_time, finish_time;
  size_t file_size = 0, per_thread_size = 0;
  char url[] = "https://test.com/test.exe";

  for (int i = 0; i <= 3; i++) {
    if (i == 3) {
      printf("get file size error!\n");
      exit(1);
    }

    if (file_size == 0) {
      file_size = get_file_size(url);
      break;
    }
  }
  per_thread_size = file_size / THREAD_COUNT;
  printf("per_thread_size=%d\n", per_thread_size);

  curr_time = get_timestamp();
  printf("start download!curr_time=%lld\n", curr_time);
  // 创建并启动线程
  for (int i = 0; i < THREAD_COUNT; i++) {
    char tmp_file[16] = {0};
    snprintf(tmp_file, sizeof(tmp_file), "test%d.exe", i);

    threadData[i].url = url;
    threadData[i].outputFilename = tmp_file;  // 应该是唯一的临时文件名
    threadData[i].startRange = i * per_thread_size;
    threadData[i].endRange = (i == (THREAD_COUNT - 1))
                                 ? (file_size - 1)
                                 : ((i + 1) * per_thread_size - 1);
    threadData[i].finish = 0;
    printf("create download thread!i=%d,file name=%s\n", i, tmp_file);
    create_thread(download_part, &threadData[i], NULL);
  }

  printf("downloading...\n");
  // 等待所有线程完成
  while (1) {
    for (int i = 0; i < THREAD_COUNT; i++) {
      if (threadData[i].finish == 0) {
        break;
      } else if (i < THREAD_COUNT - 1 && threadData[i].finish) {
        continue;
      } else if (i == THREAD_COUNT - 1 && threadData[i].finish) {
        // 所有全部下载完成
        printf("download finish!");
        goto succ;
      }
    }
    sleep(1);
  }

succ:
  finish_time = get_timestamp();
  printf("finish_time=%lld\n", finish_time);
  printf("use time = %lld\n", finish_time - curr_time);
  // 合并文件
  if (merge_files(final_output, THREAD_COUNT) != 0) {
    fprintf(stderr, "Failed to merge files\n");
    return 1;
  }

  return 0;
}

编译的时候链接上所需库：

gcc curl.c -o curltest -lcurl -lpthread

执行：

./curltest

此时查看一下是否有在下载：

pureos@pureos:~/Documents$ ls -lh | grep firmware
-rw-r--r--  1 pureos pureos 1.6M  1月  8 14:55 test0.exe
-rw-r--r--  1 pureos pureos 1.8M  1月  8 14:55 test10.exe
-rw-r--r--  1 pureos pureos 1.6M  1月  8 14:55 test11.exe
-rw-r--r--  1 pureos pureos 1.9M  1月  8 14:55 test12.exe
-rw-r--r--  1 pureos pureos 484K  1月  8 14:53 test13.exe
-rw-r--r--  1 pureos pureos 1.1M  1月  8 14:52 test14.exe
-rw-r--r--  1 pureos pureos 1.6M  1月  8 14:55 test15.exe
-rw-r--r--  1 pureos pureos 1.9M  1月  8 14:55 test1.exe
-rw-r--r--  1 pureos pureos 1.4M  1月  8 14:53 test2.exe
-rw-r--r--  1 pureos pureos 1.7M  1月  8 14:54 test3.exe
-rw-r--r--  1 pureos pureos 1.5M  1月  8 14:55 test4.exe
-rw-r--r--  1 pureos pureos 1.6M  1月  8 14:52 test5.exe
-rw-r--r--  1 pureos pureos 1.9M  1月  8 14:55 test6.exe
-rw-r--r--  1 pureos pureos 2.2M  1月  8 14:55 test7.exe
-rw-r--r--  1 pureos pureos 1.9M  1月  8 14:55 test8.exe
-rw-r--r--  1 pureos pureos 2.1M  1月  8 14:55 test9.exe

可以看到有 16 个文件，并且大小是一直在增长的。那我们就静静等待下载完成就好啦！

写在最后

通过这样一个简单的 Demo 实现了 libcurl 的多线程下载，但是在实际的业务流程中，还需要增加几点功能，提高程序的鲁棒性。如：

• 控制线程的 curl 的下载超时时间
• 增加下载失败监测机制，针对某个包下载失败可以重新单独下载
• 增加文件下载校验，如引入 MD5 校验，防止因网络传输问题导致包损坏。

参考连接

【C++】使用 libcurl 来实现多线程下载的功能