线程池使用的是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
程序输出如下,