13. 上报模块实现
程序喵
2023-12-12 08:22
阅读2274
大家好，我是程序喵。
在前面几期中，我已经介绍过如何发起http通信，如何实现加解密功能，如何进行多线程开发，这一期，我会把这些能力整合到一起，完成整个上报模块。
在写代码前，首先要梳理出上报模块需要提供什么能力。我的设计是，上报模块可以提供一个开关，只要打开开关，塞进来的消息就会每隔一段时间自动上报，同时，之前由于网络等原因上报失败的数据，也会自动从数据库中提取出来并上报。
核心点就是：
- 有开关，需要每隔一段时间自动上报
- 上报失败的数据，需要有重试
梳理后，看下对应的头文件代码：
struct BuriedData {
std::string title;
std::string data;
uint32_t priority;
};
class BuriedReportImpl;
class BuriedReport {
public:
BuriedReport(std::shared_ptr<spdlog::logger> logger,
CommonService common_service, std::string work_path);
~BuriedReport();
void Start();
void InsertData(const BuriedData& data);
private:
std::unique_ptr<BuriedReportImpl> impl_;
};
头文件代码中有几个关键点：
- BuriedData的结构定义：包含优先级、标题和详细数据，这三个由外部调用方传入具体数据
- 构造函数中的logger、common_service、work_path：因为整个SDK需要支持多实例，所以里面的各个模块，需要传递进来实际的子模块实例对象和工作目录。
- Start接口：整个上报模块的开关，调用此接口后，会自动获取数据并上报。
- InsertData接口：外部调用方想要上报某条数据就调用此接口，然后模块内部会自动挑时机上报这条数据，无需担心上报成功与否，只要调用此接口，数据就会上报成功。
- BuriedReportImpl：pimpl模式，C++中的一个很常用的设计模式，相关的实现都放在源文件中，不暴露到外边，这个模式建议大家重点学习。
然后我们再看它的实现：
BuriedReport::BuriedReport(std::shared_ptr<spdlog::logger> logger,
CommonService common_service, std::string work_path)
: impl_(std::make_unique<BuriedReportImpl>(
std::move(logger), std::move(common_service), std::move(work_path))) {
}
void BuriedReport::Start() { impl_->Start(); }
void BuriedReport::InsertData(const BuriedData& data) {
impl_->InsertData(data);
}
BuriedReport::~BuriedReport() {}
这里就是使用的pimpl模式，BuriedReport相关实现其实是调用了BuriedReportImpl的相关方法。
BuriedReportImpl类的定义如下：
static const char kDbName[] = "buried.db";
class BuriedReportImpl {
public:
BuriedReportImpl(std::shared_ptr<spdlog::logger> logger,
CommonService common_service, std::string work_path);
~BuriedReportImpl() = default;
void Start();
void InsertData(const BuriedData& data);
private:
void Init_();
void ReportCache_();
void NextCycle_();
BuriedDb::Data MakeDbData_(const BuriedData& data);
std::string GenReportData_(const std::vector<BuriedDb::Data>& datas);
bool ReportData_(const std::string& data);
private:
std::shared_ptr<spdlog::logger> logger_;
std::string work_dir_;
std::unique_ptr<BuriedDb> db_;
CommonService common_service_;
std::unique_ptr<buried::Crypt> crypt_;
std::unique_ptr<boost::asio::deadline_timer> timer_;
std::vector<BuriedDb::Data> data_caches_;
};
整个类不是只有Start和InsertData的public方法，还有好多个private方法，供类内部调用。
内部还有很多成员变量：
- logger_：打印日志的实例
- work_dir_：工作目录
- db_：数据库实例，数据库相关的方法都通过此实例调用
- common_service_：公共数据通过此实例获取
- crypt_：加解密相关的能力通过此实例调用
- timer_：定时器相关的能力使用此实例
- data_caches_：装载要上报的数据，会不停的从数据库中按优先级取数据，装载到这个cache中
构造函数
BuriedReportImpl(std::shared_ptr<spdlog::logger> logger,
CommonService common_service, std::string work_path)
: logger_(std::move(logger)),
common_service_(std::move(common_service)),
work_dir_(std::move(work_path)) {
if (logger_ == nullptr) {
logger_ = spdlog::stdout_color_mt("buried");
}
std::string key = AESCrypt::GetKey("buried_salt", "buried_password");
crypt_ = std::make_unique<AESCrypt>(key);
SPDLOG_LOGGER_INFO(logger_, "BuriedReportImpl init success");
Context::GetGlobalContext().GetReportStrand().post([this]() { Init_(); });
}
从代码中，可以看到，构造或赋值了logger和crypt对象后，通过ReportStrand调度Init_方法，看了后面的几个方法的实现后，大家就会发现整个report相关的方法都会通过ReportStrand调度，也就是说它既是在子线程中调用，又确保了所有与report相关的方法都在一个线程顺序实现，也就无需加锁处理，无需处理数据竞争问题。
Init_方法
void BuriedReportImpl::Init_() {
std::filesystem::path db_path = work_dir_;
SPDLOG_LOGGER_INFO(logger_, "BuriedReportImpl init db path: {}",
db_path.string());
db_path /= kDbName;
db_ = std::make_unique<BuriedDb>(db_path.string());
}
在此方法中，会传入数据库路径，构建db数据库对象。
Start方法
void BuriedReportImpl::Start() {
SPDLOG_LOGGER_INFO(logger_, "BuriedReportImpl start");
timer_ = std::make_unique<boost::asio::deadline_timer>(
Context::GetGlobalContext().GetMainContext(),
boost::posix_time::seconds(5));
timer_->async_wait(Context::GetGlobalContext().GetReportStrand().wrap(
[this](const boost::system::error_code& ec) {
if (ec) {
logger_->error("BuriedReportImpl::Start error: {}", ec.message());
return;
}
ReportCache_();
}));
}
在Start方法中，通过asio中的deadline_timer创建了一个定时器，并约定在5s后调用ReportCache_方法。
ps：asio中的这个定时器我个人感觉挺难用，不过这个项目中，为了保持简洁，不引入其他复杂的代码，我最终还是选择使用它，大家自己的项目中，可以考虑自己实现一个定时器和strand。
再看ReportCache_方法
void BuriedReportImpl::ReportCache_() {
SPDLOG_LOGGER_INFO(logger_, "BuriedReportImpl report cache");
if (data_caches_.empty()) {
data_caches_ = db_->QueryData(10);
}
if (!data_caches_.empty()) {
std::string report_data = GenReportData_(data_caches_);
if (ReportData_(report_data)) {
db_->DeleteDatas(data_caches_);
data_caches_.clear();
}
}
NextCycle_();
}
每次调用此方法时，会通过db对象从数据库中按优先级取10条数据，再通过GenReportData_构造网络上报的body，再调用ReportData_上报数据，上报成功后通过db对象从数据库中删除对应数据，并清空data_caches_，防止重复上报，然后调用NexeCycle_启动下一次定时任务。
如何构造网络上报的body
std::string BuriedReportImpl::GenReportData_(
const std::vector<BuriedDb::Data>& datas) {
nlohmann::json json_datas;
for (const auto& data : datas) {
std::string content =
crypt_->Decrypt(data.content.data(), data.content.size());
SPDLOG_LOGGER_INFO(logger_, "BuriedReportImpl report data content size: {}",
data.content.size());
json_datas.push_back(content);
}
std::string ret = json_datas.dump();
return ret;
}
因为数据库中拿出来的10条数据都是被加密过的，所以取出来后我们需要先进行解密操作，然后通过json构造成一个json array，最后再序列化成string，也就是网络上报的body。
如何启动下一次循环任务
void BuriedReportImpl::NextCycle_() {
SPDLOG_LOGGER_INFO(logger_, "BuriedReportImpl next cycle");
timer_->expires_at(timer_->expires_at() + boost::posix_time::seconds(5));
timer_->async_wait([this](const boost::system::error_code& ec) {
if (ec) {
logger_->error("BuriedReportImpl::NextCycle_ error: {}", ec.message());
return;
}
Context::GetGlobalContext().GetReportStrand().post(
[this]() { ReportCache_(); });
});
}
还是使用asio的定时器，约定5s后再次调用ReportCache_方法即可。
用户如何向上报模块插入数据
void BuriedReportImpl::InsertData(const BuriedData& data) {
Context::GetGlobalContext().GetReportStrand().post(
[this, data]() { db_->InsertData(MakeDbData_(data)); });
}
首先需要切线程，统一放到ReportStrand模块调度任务，因为我们也不知道外部究竟是在哪个线程调用的此方法，如果不是相同的Strand，就会产生数据竞争的问题，所以这里需要通过ReportStrand post对应的任务。
如何将数据存入数据库中
BuriedDb::Data BuriedReportImpl::MakeDbData_(const BuriedData& data) {
BuriedDb::Data db_data;
db_data.id = -1;
db_data.priority = data.priority;
db_data.timestamp = std::chrono::duration_cast<std::chrono::milliseconds>(
std::chrono::system_clock::now().time_since_epoch())
.count();
nlohmann::json json_data;
json_data["title"] = data.title;
json_data["data"] = data.data;
json_data["user_id"] = common_service_.user_id;
json_data["app_version"] = common_service_.app_version;
json_data["app_name"] = common_service_.app_name;
json_data["custom_data"] = common_service_.custom_data;
json_data["system_version"] = common_service_.system_version;
json_data["device_name"] = common_service_.device_name;
json_data["device_id"] = common_service_.device_id;
json_data["buried_version"] = common_service_.buried_version;
json_data["lifecycle_id"] = common_service_.lifecycle_id;
json_data["priority"] = data.priority;
json_data["timestamp"] = CommonService::GetNowDate();
json_data["process_time"] = CommonService::GetProcessTime();
json_data["report_id"] = CommonService::GetRandomId();
std::string report_data = crypt_->Encrypt(json_data.dump());
db_data.content = std::vector<char>(report_data.begin(), report_data.end());
SPDLOG_LOGGER_INFO(logger_, "BuriedReportImpl insert data size: {}",
db_data.content.size());
return db_data;
}
BuriedData中有用户传进来的优先级、title、data等数据，再加上common_service中的公共数据，再加上当前时间等数据，构造成一个json object，然后序列化后再加密，即可通过db对象存入数据库中。
整个上报模块需要特别注意几个关键点：
- 通过定时器启动任务，确保每隔一段时间都能上报数据
- 通过strand调度任务，确保report模块内的操作都通过一个strand调度，避免数据竞争问题
- 存数据前要加密，取数据后要解密，更安全一些
到这里，整个上报模块已经介绍完毕，下期我会具体介绍下如何获取的系统信息，下期见。
评论
问答助学
相关内容
0个评论
全部评论
点击登录，快来和大家讨论吧～
表情
图片
暂无评论
作者分享
分享最近遇到的C符号冲突问题的排查过程
7
【C++埋点SDK实战】加餐-如何在简历中体现本项目
6
#ifndef vs #pragma once
12
如何准备C++面试？
23
http://mindhacks.cn/2011/11/04/how-to-interview-a-person-for-two-years/
推荐大家阅读这篇文章《https://wx.zsxq.com/mweb/views/weread/search.html?keyword=怎样花两年时间去面试一个人》，我倒建议大家可以花两年时间成为那个优秀候选人，阅读优质书籍，开源自己的优质代码，写有足够深度的博客，让别人无需看你的简历就知道你是那个合适的人。
32
