C++11 使用线程池多线程下载图片

线程池使用的是boost::asio的thread_pool。这个肯定没问题了,方便好用,准标准库。
另外还使用了前几天封装的http_util和sf_db2库。
以及nlohmann::json库和pystring库。
其他的几个库都是在github上面现成的,或者在之前的文章里面有讲过。详情可以看链接。
http_util: https://www.jianshu.com/p/1e6c849f3be5
sf_db2: https://www.jianshu.com/p/c9472115fd0d
nlohmann::json: https://github.com/nlohmann/json
boost::asio: 这个可以参考Mac下boost库的安装。
其中有另外一个库是今天加的,
pystring。看了下源码好像是基于boost实现的。但是他的Makefile文件不好使,也就是说没法做本地安装。
我自己写了一个CMakeLists文件,用来安装pystring库。
pystring github地址,
https://github.com/imageworks/pystring
可以使用下面的CMakeLists.txt编译,安装pystring库。

cmake_minimum_required(VERSION 2.6) project(hello_world) add_definitions(-std=c++14) add_definitions(-g) include_directories(/usr/local/include ${CMAKE_CURRENT_SOURCE_DIR}) LINK_DIRECTORIES(/usr/local/lib) file( GLOB APP_SOURCES ${CMAKE_CURRENT_SOURCE_DIR}/*.cpp ${CMAKE_CURRENT_SOURCE_DIR}/*.h) foreach( sourcefile ${APP_SOURCES} ) file(RELATIVE_PATH filename ${CMAKE_CURRENT_SOURCE_DIR} ${sourcefile}) if( "${filename}" STREQUAL "pystring.cpp") string(REPLACE ".cpp" "" file ${filename}) add_library(${file} SHARED ${APP_SOURCES}) add_library(${file}_static STATIC ${APP_SOURCES}) INSTALL (TARGETS ${file} ${file}_static LIBRARY DESTINATION lib ARCHIVE DESTINATION lib) INSTALL (FILES ${file}.h DESTINATION include/${file}) endif() endforeach( sourcefile ${APP_SOURCES})

程序实现的功能主要有以下几块。
1、从Snowflake数据库拿取图片URL地址需要的json字段,并封装成std::vector<image_ele>对象
2、切割大的std::vector对象,生成子对象。好使用多线程处理。一般数据量比较大的情况下,或者耗时操作,都是这个思路。
3、对于每个子的vector对象,使用boost::asio::thread_pool开启线程池,批量下载。下载使用的是http库。

代码如下,
CMakeLists.txt

cmake_minimum_required(VERSION 2.6) project(hello_world) add_definitions(-std=c++14) add_definitions(-g) find_package(Boost REQUIRED COMPONENTS system filesystem serialization program_options thread include_directories(${Boost_INCLUDE_DIRS} /usr/local/include ${CMAKE_CURRENT_SOURCE_DIR}/../include) LINK_DIRECTORIES(/usr/local/lib) file( GLOB APP_SOURCES ${CMAKE_CURRENT_SOURCE_DIR}/*.cpp ${CMAKE_CURRENT_SOURCE_DIR}/*.h ${CMAKE_CURRENT_SOURCE_DIR}/../include/http/impl/*.cpp) foreach( sourcefile ${APP_SOURCES} ) file(RELATIVE_PATH filename ${CMAKE_CURRENT_SOURCE_DIR} ${sourcefile}) if( "${filename}" STREQUAL "main.cpp") string(REPLACE ".cpp" "" file ${filename}) add_executable(${file} ${APP_SOURCES}) target_link_libraries(${file} ${Boost_LIBRARIES}) target_link_libraries(${file} pthread ssl crypto libgtest.a libgmock.a iodbc iodbcinst libnanodbc.a pystring pthread) endif() endforeach( sourcefile ${APP_SOURCES})

main.cpp

#include "http/http_util.h"
#include "json/json.hpp"
#include "pystring/pystring.h"
#include "sf_db2/sf_db2.h"
#include <boost/asio/thread_pool.hpp>
#include <boost/asio/post.hpp>
#include <vector>
using json = nlohmann::json;
struct image_ele {
    std::string product_id;
    std::string image_path;
const int BatchSize = 1000;
const std::string ConnStr = "dsn=product_odbc;pwd={YOUR_PASSWORD}";
const std::string ImagePath = "./images";
const std::string CdnHost = "static-s.aa-cdn.net";
std::vector<image_ele> get_images_from_db() {
    auto conn_str = ConnStr;
    auto raw_query =
        R"(select product_key, screenshots_md5 
        from AA_INTELLIGENCE_PRODUCTION.ADL_MASTER.DIM_PRODUCT_LOCALIZED_DETAIL_V1_CLUSTER_BY_PRODUCT_KEY
        limit 2000;)";
    sf_connection sf{conn_str};
    auto res = sf.exec_raw_query(raw_query);
    int ele_size = res.affected_rows();
    const auto columns = res.columns();
    std::vector<image_ele> res_eles{};
    const std::string null_value = "null";
    while (res.next()) {
        auto const product_id_ = res.get<std::string>(0, null_value);
        auto const image_json_ = res.get<std::string>(1, null_value);
        auto image_obj = json::parse(image_json_);
        // 使用 nlohmann::json 库解析json对象,拿取对象中的URL
        // 对象格式 {"default":
        // ["gp/20600013289355/OpozImyAlqDxklfG2v3MSHpUfWxeCUIhz2nqJf_g9knQU2cd9o4vY7OSSUnM7ElzBDyI"]}
        for (auto &&image_it = image_obj.begin(), end = image_obj.end();
             image_it != end; ++image_it) {
            auto img_list = *image_it;
            for (auto&& image_url : img_list) {
                image_ele ele{product_id_, image_url};
                res_eles.emplace_back(std::move(ele));
    return std::move(res_eles);
bool test_a_image(const std::string& host, const std::string& path) {
    std::string final_path = "/img/" + path;
    std::string result_name = path;
    // 原先URL path替换 "/"为 "_",作为文件名,就不用自己生成UUID了
    auto tmp_result_name = pystring::replace(result_name, "/", "_");
    std::string final_result_name = ImagePath + "/" + tmp_result_name + ".png";
    bool res = HttpUtil::get_file(host, final_path, final_result_name);
    if (!res) {
        std::cerr << "Download [ https://" << host << final_path << "] failed"
                  << std::endl;
        return false;
    } else {
        return true;
void download_a_batch(const std::vector<image_ele>& sub_vec) {
    for (auto&& ele : sub_vec) {
        test_a_image(CdnHost, ele.image_path);
int main(int argc, char* argv[]) {
    // 从DB获取图片信息
    auto res = get_images_from_db();
    std::cout << "Total image count: " << res.size() << std::endl;
    // 按BatchSize大小进行分批,放进subVector中
    std::size_t batches = res.size() / BatchSize + 1;
    std::vector<std::vector<image_ele>> sub_eles;
    for (int i = 0; i < batches; ++i) {
        std::vector<image_ele> sub_ele{};
        if (i + 1 < batches) {
            for (int j = 0; j < BatchSize; ++j) {
                sub_ele.emplace_back(std::move(res[i * BatchSize + j]));
        } else {
            for (int j = 0; j < res.size() % BatchSize; ++j) {
                sub_ele.emplace_back(std::move(res[i * BatchSize + j]));
        sub_eles.emplace_back(std::move(sub_ele));
    // 使用asio thread_pool启动线程池,运行子任务
    boost::asio::thread_pool pool{batches};
    for (int i = 0; i < sub_eles.size(); ++i) {
        boost::asio::post(pool,
                          std::bind(download_a_batch, std::ref(sub_eles[i])));
    pool.join();
    return 0;

http/http_util.h

#ifndef _HTTP_UTIL_H_
#define _HTTP_UTIL_H_
#define CPPHTTPLIB_OPENSSL_SUPPORT
#include "http/httplib.h"
#include <string>
class HttpUtil {
    public:
         * HttpUtil get method
         * @param: url the url to be used to get a web page from remote server
         * @param: path the path to be used to get a web page from remote server
         * @param: result_name the download result file path
        static bool get(std::string url, std::string path, std::string result_name);
         * HttpUtil get_file method
         * @param: host the host to be used to get an item from remote server
         * @param: path the path to be used to get an item from remote server
         * @param: result_name the download result file path
        static bool get_file(std::string host, std::string path, std::string result_name);
        static bool get_str(std::string host, std::string path, const std::map<std::string, std::string> & headers, std::string &result_string);
#endif

http/impl/http_util.cpp

#include "http/http_util.h"
#include <iostream>
#include <fstream>
bool HttpUtil::get(std::string url, std::string path, std::string result_name) {
    try {
        httplib::Client cli {url};
        auto res = cli.Get(path.c_str());
        if(res->status != 200) {
            std::cerr << "Get [" << url << path << "] failed" << std::endl;
            std::cerr << "Status code : [" << res->status << "]" << "\n"   << "Result body : [" << res->body << "]" 
            << std::endl; 
            return false;
        std::ofstream os {result_name, std::ios_base::out | std::ios_base::binary};
        os << res->body;
    } catch(const std::exception & e) {
        std::cerr << "Exception: " << e.what() << std::endl;
        return false;
    return true;
bool HttpUtil::get_file(std::string host, std::string path, std::string result_name) {
    try {
        #ifdef CPPHTTPLIB_OPENSSL_SUPPORT
            auto port = 443;
            httplib::SSLClient cli(host, port);
        #else
            auto port = 80;
            httplib::Client cli(host, port);
        #endif
        cli.set_connection_timeout(2);
        std::ofstream os {result_name};
        auto res = cli.Get(path.c_str(),
              [&](const char *data, size_t data_length) {
                os << std::string(data, data_length);
                return true;
        if(!res || res->status != 200) {
            return false;
    } catch(const std::exception & e) {
        std::cerr << "Exception: " << e.what() << std::endl;
        return false;
    return true;
bool HttpUtil::get_str(std::string host, std::string path, const std::map<std::string, std::string> & headers, std::string &result_string) {
     try {
        #ifdef CPPHTTPLIB_OPENSSL_SUPPORT
            auto port = 443;
            httplib::SSLClient cli(host, port);
        #else
            auto port = 80;
            httplib::Client cli(host, port);
        #endif
        cli.set_connection_timeout(2);
        httplib::Headers headers_ {};
        for(auto&& item: headers) {
            headers_.insert(item);
        auto res = cli.Get(path.c_str(), headers_);
        result_string = res->body;
        return true;
    } catch(const std::exception & e) {
        std::cerr << "Exception: " << e.what() << std::endl;
        return false;

json.hpp是源库,代码比较长,这里不贴了。
pystring.h也是源库,不贴代码了。
sf_db2/sf_db2.h

#ifndef _FREDRIC_SF_DB2_H_
#define _FREDRIC_SF_DB2_H_
#include "nanodbc/convert.h"
#include "nanodbc/nanodbc.h"
#include <any>
#include <exception>
#include <iostream>
#include <string>
using db_result = nanodbc::result;
struct sf_connection {
    nanodbc::connection conn_;
    sf_connection(const std::string& conn_str) {
        conn_ = nanodbc::connection{convert(conn_str)};
    db_result exec_raw_query(const std::string& raw_query) {
        auto res = execute(conn_, NANODBC_TEXT(raw_query));
        return std::move(res);
    template <typename... Params>
    db_result exec_prpare_statement(const std::string& pre_stmt,
                                    Params... params) {
        nanodbc::statement statement(conn_);
        std::cout << pre_stmt << std::endl;
        prepare(statement, NANODBC_TEXT(pre_stmt));
        int index = 0;
        int bind_arr[] = {(bind(statement, index, params), ++index)...};
        auto res = execute(statement);
        return std::move(res);
    virtual ~sf_connection() {}
   private:
    template <typename T>
    void bind(nanodbc::statement& stmt, int index, T param) {
        std::vector<std::string> v{std::to_string(param)};
        stmt.bind_strings(index, v);
    void bind(nanodbc::statement& stmt, int index, const char* param) {
        stmt.bind(index, param);
    void bind(nanodbc::statement& stmt, int index, std::string param) {
        stmt.bind(index, param.c_str());
#endif

程序输出如下,