手撕基于AMQP协议的简易消息队列-3（项目所用到的工具类的编写）
前置工具类的编写
基于C++11的异步操作的线程池的编写
异步操作类future的介绍
-
std::future：
-
介绍：std::future是C++11标准库中的⼀个模板类，它表⽰⼀个异步操作的结果。当我们在多线程编程中使⽤异步任务时，std::future可以帮助我们在需要的时候获取任务的执⾏结果。std::future的⼀个重要特性是能够阻塞当前线程，直到异步操作完成，从⽽确保我们在获取结果时不会遇到未完成的操作。
-
应用场景：
-
异步任务： 当我们需要在后台执⾏⼀些耗时操作时，如⽹络请求或计算密集型任务等，std::future可以⽤来表⽰这些异步任务的结果。通过将任务与主线程分离，我们可以实现任务的并⾏处理，从⽽提⾼程序的执⾏效率
-
并发控制： 在多线程编程中，我们可能需要等待某些任务完成后才能继续执⾏其他操作。通过使⽤std::future，我们可以实现线程之间的同步，确保任务完成后再获取结果并继续执⾏后续操作
-
结果获取：std::future提供了⼀种安全的⽅式来获取异步任务的结果。我们可以使⽤std::future::get()函数来获取任务的结果，此函数会阻塞当前线程，直到异步操作完成。这样，在
调⽤get()函数时，我们可以确保已经获取到了所需的结果
-
-
三种搭配future执行异步操作的使用示例
-
使⽤ std::async关联异步任务
-
std::async是⼀种将任务与std::future关联的简单⽅法。它创建并运⾏⼀个异步任务，并返回⼀个与该任务结果关联的std::future对象。默认情况下，std::async是否启动⼀个新线程，或者在等待future时，任务是否同步运⾏都取决于你给的 参数。
-
std::launch类型：
- std::launch::deferred 表明该函数会被延迟调⽤，直到在future上调⽤get()或者wait()才会开始执⾏任务
- std::launch::async 表明函数会在⾃⼰创建的线程上运⾏
- std::launch::deferred | std::launch::async 内部通过系统等条件⾃动选择策略
-
示例代码(async.cpp)
-
#include <future> // 异步操作库 #include <iostream> // 标准输入输出 #include <thread> // 线程支持库 #include <unistd.h> // POSIX系统调用（如sleep）/*** 加法函数（模拟耗时操作）* @param num1 第一个加数* @param num2 第二个加数* @return 两数之和* 注：通过sleep模拟长时间计算过程*/ int Add(int num1, int num2) {std::cout << "加法函数开始执行...\n"; // 阶段1提示sleep(20); // 模拟20秒耗时操作std::cout << "加法函数即将返回结果...\n"; // 阶段2提示return num1 + num2; }int main() {std::cout << "--------主程序开始----------\n";// 创建异步任务（延迟启动模式）// std::launch::deferred 表示延迟执行，直到调用get()/wait()时才运行// 对比：std::launch::async 会立即在独立线程执行std::future<int> result = std::async(std::launch::deferred, Add, 12, 13);sleep(10); // 主线程休眠10秒（此时异步任务尚未启动）std::cout << "--------准备获取结果----------\n";// 获取异步任务结果（此时才真正执行Add函数）// get()会阻塞直到任务完成int sum = result.get(); std::cout << "计算结果: " << sum << std::endl;std::cout << "--------程序结束----------\n";return 0; }- 编译
g++ -o async async.cpp -std=c++11 -lpthread-
结果
-
当std::launch::deferred时
-
当std::launch::async时
-
-
使⽤std::packaged_task和std::future配合
-
std::packaged_task就是将任务和 std::feature 绑定在⼀起的模板，是⼀种对任务的封装。我们可以通过std::packaged_task对象获取任务相关联的std::feature对象，通过调⽤get_future()⽅法获得。std::packaged_task的模板参数是函数签名
-
可以把std::future和std::async看成是分开的， ⽽ std::packaged_task则是⼀个整体。
-
示例代码(package_task.cpp)
#include <future> // 引入future库，用于处理异步操作 #include <iostream> // 引入输入输出流库 #include <thread> // 引入线程库// 定义一个普通的加法函数 int add(int x, int y) {return x + y; }int main() {// 创建一个packaged_task对象，它包装了add函数std::packaged_task<int(int, int)> packTask(add);// 从packaged_task对象获取一个future对象，用于访问异步操作的结果std::future<int> fu = packTask.get_future();// 创建一个线程，执行packaged_task对象。注意，我们需要传递packaged_task的引用给线程函数，// 但由于std::thread的构造函数会 其参数，而packaged_task是不可 的，所以我们必须传递一个右值引用。// 使用std::move可以将packTask转换为右值引用，从而允许线程函数接收并执行它。std::thread th1(std::move(packTask), 520, 1314);// 在主线程中等待异步操作完成，并获取结果int res = fu.get();// 打印结果std::cout << res << std::endl;// 等待线程执行完毕th1.join();return 0; }-
编译
g++ -o package_task package_task.cpp -std=c++11 -lpthread -
结果
-
-
-
使⽤std::promise和std::future配合
-
std::promise提供了⼀种设置值的⽅式，它可以在设置之后通过相关联的std::future对象进⾏读取。
-
换种说法就是之前说过std::future可以读取⼀个异步函数的返回值了， 但是要等待就绪， ⽽std::promise就提供⼀种 ⽅式⼿动让 std::future就绪
-
示例代码(promise.cpp)
#include <future> // 引入future库，用于处理异步操作 #include <iostream> // 引入输入输出流库 #include <thread> // 引入线程库// 定义一个函数，用于计算两个整数的和，并通过promise对象返回结果 void add(int x, int y, std::promise<int> &prom) {std::cout << "add function" << std::endl; // 打印信息，表示add函数被调用prom.set_value(x + y); // 设置promise的值，这个值会被future对象获取 }int main() {std::promise<int> prom; // 创建一个promise对象，用于存储类型为int的结果std::future<int> fut = prom.get_future(); // 从promise对象获取一个future对象，用于访问结果// 创建一个线程，执行add函数，传入两个整数和一个promise对象的引用std::thread th1(add, 520, 1314, std::ref(prom));// 在主线程中获取add函数的计算结果int res = fut.get(); // 调用future对象的get方法，获取结果，并等待异步操作完成std::cout << res << std::endl; // 打印结果th1.join(); // 等待线程th1执行完毕return 0; } -
编译
g++ -o promise promise.cpp -std=c++11 -lpthread -
结果
-
-
-
C++11线程池的实现
-
异步操作的搭配选择：基于线程池执⾏任务的时候，⼊⼝函数内部执⾏逻辑是固定的，因此选择std::packaged_task加上std::future的组合来实现
-
线程池的工作思想：⽤⼾传⼊要执⾏的函数，以及需要处理的数据（函数的参数）,由线程池中的⼯作线程来执⾏函数完成任务
-
线程池需要管里的成员：
- 任务池：⽤vector维护的⼀个函数任务池⼦
- 互斥锁 & 条件变量： 实现同步互斥
- ⼀定数量的⼯作线程：⽤于不断从任务池取出任务执⾏任务
- 结束运⾏标志：以便于控制线程池的结束
-
线程池需要管理的操作
- ⼊队任务：⼊队⼀个函数和参数
- 停⽌运⾏：终⽌线程池
-
线程池的实现代码(ThreadPool.cpp)
#include <condition_variable> #include <functional> #include <future> #include <iostream> #include <memory> #include <mutex> #include <thread> #include <vector>class ThreadPool { public:// 定义任务类型为无参数无返回值的函数using Task = std::function<void(void)>;// 构造函数，初始化线程池// 参数为线程池中的线程数量，默认为1// 初始化停止标志为falseThreadPool(int thr_count = 1) : _stop(false){for (int i = 0; i < thr_count; i++){_threads.emplace_back(&ThreadPool::Entry, this);}}// 析构函数，用于清理线程池// 停止所有线程并等待它们完成~ThreadPool(){Stop();}// 模板函数，用于向线程池中添加任务// 返回一个future，包含任务的返回值template <typename F, typename... Args>auto push(const F &&func, Args &&...args)-> std::future<decltype(func(args...))>{//无参函数对象->packaged_task->任务对象智能指针// 将传入的函数及函数的参数绑定为一个无参函数对象，之后再构造一个智能指针去管理一个packaged_task对象，最后封装为任务指针，放入任务池//封装为任务指针是因为packaged_task并不是一个函数对象,不可以直接传给线程执行using return_type = decltype(func(args...));auto function = std::bind(std::forward<F>(func), std::forward<Args>(args)...);//参数包的展开auto task = std::make_shared<std::packaged_task<return_type()>>(function);std::future<return_type> fu = task->get_future();{std::unique_lock<std::mutex> uniLock(_mutex);// 将任务添加到任务队列中,这里添加进去的是lamda函数对象，并不是task指针_tasks.emplace_back([task](){ (*task)(); });// 通知一个等待线程有任务可处理_conv.notify_one();}return fu;}// 停止线程池中的所有线程bool Stop(){// 如果已经停止，则直接返回trueif (_stop == true)return true;_stop = true;// 通知所有等待线程停止_conv.notify_all();// 等待所有线程完成for (auto &th : _threads){th.join();}return true;}private:void Entry(){// 当_stop标志为false时，表示线程池仍在运行，持续处理任务while (false == _stop){std::vector<Task> tmp_tasks; // 创建一个临时任务池，用于存储待处理的任务{// 加锁，保护对共享资源（如_tasks和_stop）的访问std::unique_lock<std::mutex> lock(_mutex);// 等待条件满足：任务池不为空，或者收到停止信号_conv.wait(lock, [this](){ return _stop || !_tasks.empty(); });// 条件满足后，将_tasks中的任务转移到临时任务池tmp_tasks中// 这样做是为了避免在持有锁的情况下执行任务，从而提高效率tmp_tasks.swap(_tasks);}// 锁已经释放，现在可以安全地执行所有取出的任务for (auto &tmp_task : tmp_tasks){tmp_task(); // 执行任务}}// 当_stop为true时，退出循环，函数返回return;}private:std::mutex _mutex; // 互斥锁，保护任务和停止标志std::vector<Task> _tasks; // 存储待处理的任务队列std::condition_variable _conv; // 条件变量，用于线程间的同步和通知std::atomic<bool> _stop; // 原子标志，指示线程池是否应停止std::vector<std::thread> _threads; // 存储线程池中的所有工作线程 };int Add(int num1, int num2) {return num1 + num2; }int main() {ThreadPool pool;for (int i = 0; i < 10; i++) {std::future<int> fu = pool.push(Add, 11, i);std::cout << fu.get() << std::endl;}pool.Stop();return 0; } -
编译
g++ -o ThreadPool ThreadPool.cpp -std=c++11 -lpthread -
结果
日志打印工具
-
为了便于编写项⽬中能够快速定位程序的错误位置，因此编写⼀个⽇志打印类，进⾏简单的⽇志打印。
-
实现代码
#ifndef __M_LOG_H__ // 防止头文件重复包含 #define __M_LOG_H__#include <ctime> // 时间相关函数 #include <iostream> // 标准输入输出// ==================== 日志级别定义 ==================== #define DBG_LEVEL 0 // 调试级别（最低级别） #define INF_LEVEL 1 // 信息级别 #define ERR_LEVEL 2 // 错误级别（最高级别） #define DEFAULT_LEVEL DBG_LEVEL // 默认输出级别（当前设为调试级别）// ==================== 核心日志宏 ==================== /*** @brief 通用日志输出宏（内部使用）* @param lev_str 级别字符串（如"DBG"）* @param level 日志级别数值* @param format 格式化字符串（类似printf）* @param ... 可变参数（对应format中的占位符）* 功能：根据日志级别决定是否输出，带时间戳和代码位置信息*/ #define LOG(lev_str, level, format, ...) \{ \if (level >= DEFAULT_LEVEL) /* 级别过滤 */ \{ \time_t tmp = time(nullptr); /* 获取当前时间戳 */ \struct tm *time = localtime(&tmp); /* 转换为本地时间 */ \char str_time[32]; /* 时间格式化缓冲区 */ \size_t ret = strftime(str_time, 31, "%D %H:%M:%S", time); /* 格式化为 MM/DD/YY HH:MM:SS */ \printf("[%s][%s][%s:%d]\t" format "\n", /* 输出格式： \lev_str, /* 日志级别标记 */ \str_time, /* 格式化时间 */ \__FILE__, /* 源代码文件名 */ \__LINE__, /* 代码行号 */ \##__VA_ARGS__); /* 用户日志内容 */ \} \}// ==================== 快捷日志宏 ==================== /// 调试日志（DBG_LEVEL） #define DLOG(format, ...) LOG("DBG", DBG_LEVEL, format, ##__VA_ARGS__) /// 信息日志（INF_LEVEL） #define ILOG(format, ...) LOG("INF", INF_LEVEL, format, ##__VA_ARGS__) /// 错误日志（ERR_LEVEL） #define ELOG(format, ...) LOG("ERR", ERR_LEVEL, format, ##__VA_ARGS__)#endif // __M_LOG_H__
文件基础操作类
-
需要提供的操作：
-
⽂件是否存在判断
-
⽂件⼤⼩获取
-
⽂件读/写
-
⽂件创建/删除
-
⽬录创建/删除
-
-
实现代码
//文件操作类 class FileHelper {public:// 构造函数FileHelper(const std::string &filename): _filename(filename){}// 文件大小size_t size(){struct stat st;int ret = stat(_filename.c_str(), &st);if (ret != 0)return 0;return st.st_size;}// 是否存在bool exists(){struct stat st;return stat(_filename.c_str(), &st) == 0;}// 读操作bool read(char *body, size_t offset, size_t len){// 打开文件std::ifstream ifs(_filename, std::ios::binary | std::ios::in); // 以二进制形式打开if (ifs.is_open() == false){ELOG("%s 文件打开失败！", _filename.c_str());return false;}// 跳转到指定位置ifs.seekg(offset, std::ios::beg);// 读取文件ifs.read(body, len);if (ifs.good() == false){ELOG("%s 文件读取数据失败！！", _filename.c_str());ifs.close();return false;}// 关闭文件ifs.close();return true;}bool read(std::string &body){// 获取文件大小，根据文件大小调整body的空间size_t fsize = this->size();body.resize(fsize);return read(&body[0], 0, fsize);}// 写操作bool write(const char *body, size_t offset, size_t len){// 打开文件std::fstream fs(_filename, std::ios::binary | std::ios::in | std::ios::out);if (fs.is_open() == false){ELOG("%s 文件打开失败！", _filename.c_str());return false;}// 跳转到文件指定位置fs.seekp(offset, std::ios::beg);// 写入数据fs.write(body, len);if (fs.good() == false){ELOG("%s 文件写入数据失败！！", _filename.c_str());fs.close();return false;}// 关闭文件fs.close();return true;}bool write(const std::string &body){//如果文件有内容，则删除源文件，重新生成名为filename的文件if (size() > 0){removeFile(_filename);FileHelper::createFile(_filename);}return write(body.c_str(), 0, body.size());}// 重命名bool rename(const std::string &newfilename){return (::rename(_filename.c_str(), newfilename.c_str()) == 0);}// 获取父目录static std::string parentDirectory(const std::string &filename){int pos = filename.find_last_of("/");if (pos == std::string::npos){return "./";}std::string ppath = filename.substr(0, pos);return ppath;}// 创建文件static bool createFile(const std::string &filename){std::fstream ofs(filename, std::ios::out | std::ios::binary);if (!ofs.is_open()){ELOG("创建文件：%s 失败！", filename.c_str());return false;}ofs.close();return true;}// 删除文件static bool removeFile(const std::string &filename){return (::remove(filename.c_str()) == 0);}// 创建目录static bool createDirectory(const std::string &path){int pos = 0, idx = 0;while (idx < path.size()){pos = path.find("/", idx);if (pos == std::string::npos){return (::mkdir(path.c_str(), 0775) == 0);}std::string subpath = path.substr(0, pos);int ret = ::mkdir(subpath.c_str(), 0775);if (ret != 0 && errno != EEXIST){ELOG("创建目录 %s 失败: %s", subpath.c_str(), strerror(errno));return false;}idx = pos + 1;}return true;}// 删除目录static bool removeDirectory(const std::string &path){std::string command = "rm -rf " + path;return (::system(command.c_str()) != -1);}private:std::string _filename; };
SQLite基础操作类
-
需要提供的操作：
- 判断库是否存在
- 创建并打开/关闭/删除库
- 启动/提交/回滚事物
- 执行语句
-
实现代码
class SqliteHelper { public:// 定义一个回调函数类型，用于sqlite3_exec的回调typedef int (*SqliteCallback)(void *, int, char **, char **);SqliteHelper(){}// 构造函数，接收数据库文件名SqliteHelper(const std::string dbfilename): _dbfilename(dbfilename){}// 打开数据库// 参数safe_leve用于指定打开数据库的附加模式，默认为SQLITE_OPEN_FULLMUTEXbool open(int safe_leve = SQLITE_OPEN_FULLMUTEX){// 使用sqlite3_open_v2函数打开或创建数据库int ret = sqlite3_open_v2(_dbfilename.c_str(), &_handler, SQLITE_OPEN_READWRITE | SQLITE_OPEN_CREATE | safe_leve, nullptr);if (ret != SQLITE_OK){// std::cout << "创建/打开sqlite数据库失败: ";// std::cout << sqlite3_errmsg(_handler) << std::endl;ELOG("创建/打开sqlite数据库失败: %s", sqlite3_errmsg(_handler));return false;}return true;}// 关闭数据库void close(){// 使用sqlite3_close_v2函数关闭数据库if (_handler)sqlite3_close_v2(_handler);}// 执行SQL语句// 参数sql为要执行的SQL语句，cb为回调函数，arg为回调函数的参数bool exec(const std::string &sql, SqliteCallback cb, void *arg){// 使用sqlite3_exec函数执行SQL语句int ret = sqlite3_exec(_handler, sql.c_str(), cb, arg, nullptr);if (ret != SQLITE_OK){// std::cout << sql << std::endl;// std::cout << "执行语句失败: ";// std::cout << sqlite3_errmsg(_handler) << std::endl;ELOG("执行语句：%s 失败!\t错误原因: %s", sql.c_str(), sqlite3_errmsg(_handler));return false;}return true;}private:std::string _dbfilename; // 数据库文件名sqlite3 *_handler; // 数据库句柄 };
字符串操作类
-
一共字符串分割功能
-
实现代码
class StrHelper { public:// 静态成员函数，用于分割字符串// 参数：待分割的字符串str，分隔符sep，存储结果的向量result// 返回值：分割后的子字符串数量static size_t split(const std::string &str, const std::string &sep, std::vector<std::string> &result){int pos = 0; // 用于记录分隔符在字符串中的位置int idx = 0; // 用于遍历字符串的索引// 遍历字符串，直到索引超出字符串长度while (idx < str.size()){// 从当前索引开始查找分隔符，返回分隔符的起始位置pos = str.find(sep, idx);// 如果未找到分隔符，说明已经到达字符串末尾或分隔符不在剩余部分if (pos == std::string::npos){// 将剩余部分作为最后一个子字符串添加到结果向量中std::string tmp = str.substr(idx);result.push_back(tmp);// 返回结果向量的大小，即子字符串的数量return result.size();}// 如果分隔符的起始位置与当前索引相同，说明两个分隔符之间没有数据if (pos == idx){// 更新索引，跳过这个分隔符idx = pos + sep.size();continue;}// 提取两个分隔符之间的子字符串，并添加到结果向量中std::string tmp = str.substr(idx, pos - idx);result.push_back(tmp);// 更新索引，跳过这个分隔符和已经处理的子字符串idx = pos + sep.size();}// 返回结果向量的大小，即子字符串的数量return result.size();} };
UUID⽣成器类
-
UUID(Universally Unique Identifier), 也叫通⽤唯⼀识别码，通常由32位16进制数字字符组成。
-
在本项目中，uuid⽣成，我们采⽤⽣成8个随机数字，加上8字节序号，共16字节数组⽣成32位16进制字符的组合形式来确保全局唯⼀的同时能够根据序号来分辨数据
-
实现代码
/*** @class UUIDHelper* @brief UUID生成工具类（伪随机实现）* * 生成格式：xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx* 实现特点：* 1. 前16字符为随机数（std::mt19937_64生成）* 2. 后16字符为递增序列号（原子操作保证线程安全）* 3. 16进制表示，小写字母* 注意：非标准UUID实现，不适合高安全场景*/ class UUIDHelper { public:/*** @brief 生成伪随机UUID字符串* @return 格式化的UUID字符串（36字符）* * 生成逻辑：* 1. 前16字符：8个随机字节（显示为16个十六进制字符）* 2. 中间3个连字符"-"* 3. 后16字符：8字节递增序列号（显示为16个十六进制字符）* 4. 最后1个连字符"-"*/static std::string uuid(){// 初始化随机数生成器std::random_device rd; // 真随机数种子std::mt19937_64 gernator(rd()); // 64位梅森旋转算法std::uniform_int_distribution<int> distribution(0, 255); // 生成0-255的随机数std::stringstream ss;// 生成前16位随机部分（8字节）for (int i = 0; i < 8; i++){// 格式化为2位十六进制（不足补零）ss << std::setw(2) << std::setfill('0') << std::hex << distribution(gernator);// 在特定位置插入分隔符if (i == 3 || i == 5 || i == 7) // 第4、6、8字节后{ss << "-";}}// 生成后16位序列号部分（原子计数器）static std::atomic<size_t> seq(1); // 静态原子计数器size_t num = seq.fetch_add(1); // 原子自增// 将序列号转为十六进制字符串for (int i = 7; i >= 0; i--) // 从高字节到低字节{// 每次处理1字节（2个十六进制字符）ss << std::setw(2) << std::setfill('0') << std::hex << ((num >> (i * 8)) & 0xff); // 提取字节// 在倒数第2字节后插入分隔符if (i == 6)ss << "-";}return ss.str(); // 示例：3e2a5f8c-d4b1-47e2-9a3b-000000000001} };/* 典型输出格式（示例）：前半随机部分 分隔 递增序列号┌───────────┐ ┬ ┌───────────┐3e2a5f8c-d4b1-47e2-9a3b-000000000001│ └── 序列号从1开始└─ 固定位置的分隔符 */