当前位置： 首页 > java >正文

C++ - 仿 RabbitMQ 实现消息队列--服务端核心模块实现（一）

java 2025/7/21 8:48:49

目录

日志打印工具

实用 Helper 工具

sqlite 基础操作类

 字符串操作类

 UUID 生成器类

 文件基础操作

文件是否存在判断

文件大小获取

读文件

写文件

文件重命名

文件创建/删除

 父级目录的获取

目录创建/删除

附录（完整代码）

日志打印工具

        为了便于编写项目中能够快速定位程序的错误位置，因此编写一个日志输出宏工具，进行简单的日志打印。

  • 定义DEBUG，INFO，ERROR三个日志等级。
  • 日志格式：[日志等级][时间][文件][行号]    日志信息。
#pragma once#include <iostream>
#include <ctime>
#include <string>namespace jiuqi
{
#define DEBUG_LEVEL 0
#define INFO_LEVEL 1
#define ERROR_LEVEL 2#define DEFAULT_LEVEL DEBUG_LEVEL#define LOG(level, fmt, ...)                                                                                \{                                                                                                         \if (level >= DEFAULT_LEVEL)                                                                           \{                                                                                                     \std::string level_str;                                                                            \if (level == DEBUG_LEVEL)                                                                         \level_str = "DEBUG";                                                                          \else if (level == INFO_LEVEL)                                                                     \level_str = "INFO";                                                                           \else                                                                                              \level_str = "ERROR";                                                                          \time_t t = time(nullptr);                                                                         \struct tm *tm = localtime(&t);                                                                    \char time[32];                                                                                    \strftime(time, 31, "%H:%M:%S", tm);                                                               \printf("[%s][%s][%s:%d]\t" fmt "\n", level_str.c_str(), time,  __FILE__, __LINE__, ##__VA_ARGS__);\}                                                                                                     \}#define DEBUG(fmt, ...) LOG(DEBUG_LEVEL, fmt, ##__VA_ARGS__)
#define INFO(fmt, ...) LOG(INFO_LEVEL, fmt, ##__VA_ARGS__)
#define ERROR(fmt, ...) LOG(ERROR_LEVEL, fmt, ##__VA_ARGS__)
}

实用 Helper 工具

        Helper 工具类中要完成的是项目中需要的一些辅助零碎的功能代码实现，其中包括文
件的基础操作，字符串的额外操作等在项目中用到的零碎功能。

sqlite 基础操作类

        使用我们之前demo中编写过的sqlite类即可，将输出替换为日志打印。

    class SqliteHelper{public:typedef int (*sqliteCallback)(void *, int, char **, char **);SqliteHelper(const std::string &dbfile): _dbfile(dbfile), _handler(nullptr) {}bool open(int safe_level = SQLITE_OPEN_FULLMUTEX){int ret = sqlite3_open_v2(_dbfile.c_str(), &_handler, SQLITE_OPEN_READWRITE | SQLITE_OPEN_CREATE | safe_level, nullptr);if (ret != SQLITE_OK){// std::cerr << "打开/创建数据库失败: " << sqlite3_errmsg(_handler) << std::endl;ERROR("%s:%s", "打开/创建数据库失败", sqlite3_errmsg(_handler));return false;}return true;}bool exec(const std::string &sql, sqliteCallback callback, void *arg){int ret = sqlite3_exec(_handler, sql.c_str(), callback, arg, nullptr);if (ret != SQLITE_OK){// std::cerr << sql << std::endl;// std::cerr << "执行语句失败: " << sqlite3_errmsg(_handler) << std::endl;ERROR("%s\n%s:%s", sql.c_str(), "执行语句失败: ", sqlite3_errmsg(_handler));return false;}return true;}void close(){if (_handler)sqlite3_close_v2(_handler);}private:std::string _dbfile;sqlite3 *_handler;};

 字符串操作类

        这里先实现字符串分割功能，主要是用于后续消息类型与队列类型，即绑定信息匹配的问题。

    class StrHelper{public:static size_t split(const std::string &str, const std::string &sep, std::vector<std::string> &result){size_t end = 0, start = 0;while (start < str.size()){end = str.find(sep, start);if (end == std::string::npos){result.push_back(str.substr(start));break;}if (end != start){result.push_back(str.substr(start, end - start));}start = end + sep.size();}return result.size();}};

说明：

  • 以sep作为分割符将str分割，结果存放在result中。
  • 返回分割子串的数量。
  • 注意，如果出现连续的分割符，我们必须跳过，即不允许插入空串。

 UUID 生成器类

        UUID(Universally Unique Identifier), 也叫通用唯一识别码，通常由 32 位 16 进制数字字符组成。

        在这里，uuid 生成，我们采用生成 8 个随机数字，加上 8 字节序号，共 16 字节数组生成 32 位 16 进制字符的组合形式来确保全局唯一的同时能够根据序号来分辨数据。

        为了获得随机数与稳定递增的序号（需支持高并发），我们介绍一下几个对象：

  • 随机数：

    • std::random_device对象：生成机器随机数，真随机数，但是效率较低。

    • std::mt19937_64对象：64位梅森旋转算法（Mersenne Twister）生成伪随机数，可以使用std::random_device对象生成的随机数作为种子。

    • std::uniform_int_distribution<int>对象：将随机数引擎的输出转换为均匀分布的整数。

  • 序号：
    • 使用size_t的原子计数器，使用fetch_add接口可以获得稳定递增的序号。
    class UUIDHelper{public:static std::string uuid(){std::random_device rd;std::mt19937_64 gernator(rd());std::uniform_int_distribution<int> distribution(0, 255);std::stringstream ss;for (int i = 1; i <= 8; i++){ss << std::setw(2) << std::setfill('0') << std::hex << distribution(gernator);if (i % 2 == 0)ss << '-';}static std::atomic<size_t> sep(1);size_t num = sep.fetch_add(1);for (int i = 7; i >= 0; i--){ss << std::setw(2) << std::setfill('0') << std::hex << ((num >> (i * 8)) & 0xff);if (i == 4)ss << '-';}return ss.str();}};

 文件基础操作

        我们将提供一系列文件操作，每个功能实现静态和非静态操作两种。

文件是否存在判断

        bool exists(){struct stat st;return (stat(_filename.c_str(), &st) == 0);}static bool exists(const std::string &filename){struct stat st;return (stat(filename.c_str(), &st) == 0);}

        使用stat接口可跨平台。

文件大小获取

        size_t size(){struct stat st;int ret = stat(_filename.c_str(), &st);if (ret < 0)return 0;return st.st_size;}static size_t size(const std::string &filename){struct stat st;int ret = stat(filename.c_str(), &st);if (ret < 0)return 0;return st.st_size;}

        同样使用stat接口，文件大小会被存放在结构体st中。

读文件

        bool read(std::string &body){size_t fsize = this->size();return read(body, 0, fsize);}bool read(std::string &body, size_t offset, size_t len){// 打开文件std::ifstream ifs(_filename, std::ios::binary | std::ios::in);if (!ifs.is_open()){ERROR("%s:文件打开失败", _filename.c_str());return false;}// 移动文件读取位置ifs.seekg(offset, std::ios::beg);// 读取文件body.resize(len);ifs.read(&body[0], len);if (!ifs.good()){ERROR("%s:文件读取失败", _filename.c_str());ifs.close();return false;}// 关闭文件返回ifs.close();return true;}static bool read(const std::string &filename, std::string &body){size_t fsize = size(filename);return read(filename, body, 0, fsize);}static bool read(const std::string &filename, std::string &body, size_t offset, size_t len){// 打开文件std::ifstream ifs(filename, std::ios::binary | std::ios::in);if (!ifs.is_open()){ERROR("%s:文件打开失败", filename.c_str());return false;}// 移动文件读取位置ifs.seekg(offset, std::ios::beg);// 读取文件body.resize(len);ifs.read(&body[0], len);if (!ifs.good()){ERROR("%s:文件读取失败", filename.c_str());ifs.close();return false;}// 关闭文件返回ifs.close();return true;}

        我们提供读取文件全部内容与读取文件特定位置特定长度内容的接口，有两点需要注意：

  1. 每次读取需要把读取缓冲区（即body）resize，否则会导致读取失败。
  2. ifstream对象的read接口第一个参数是char*类型的，不能使用string对象的c_str接口（因为返回的是const char*），解决办法是传入body第一个字符的地址。

写文件

        bool write(const std::string &body){size_t fsize = this->size();return write(body.c_str(), fsize, body.size());}bool write(const char *body, size_t offset, size_t len){// 打开文件std::fstream fs(_filename, std::ios::binary | std::ios::out | std::ios::in);if (!fs.is_open()){ERROR("%s:文件打开失败", _filename.c_str());return false;}// 移动文件写入位置fs.seekp(offset, std::ios::beg);// 写入文件fs.write(body, len);if (!fs.good()){ERROR("%s:文件读取失败", _filename.c_str());fs.close();return false;}fs.close();return true;}static bool write(const std::string &filename, const std::string &body){size_t fsize = size(filename);return write(filename, body.c_str(), fsize, body.size());}static bool write(const std::string &filename, const char *body, size_t offset, size_t len){// 打开文件std::fstream fs(filename, std::ios::binary | std::ios::out | std::ios::in);if (!fs.is_open()){ERROR("%s:文件打开失败", filename.c_str());return false;}// 移动文件写入位置fs.seekp(offset, std::ios::beg);// 写入文件fs.write(body, len);if (!fs.good()){ERROR("%s:文件读取失败", filename.c_str());fs.close();return false;}fs.close();return true;}

         这里实现了追加写入与在特定位置写入的功能，但是在特定位置写入会覆盖原本的内容，暂时未实现真正的插入功能。

        需要注意，因为要移动文件写位置，必须以读写方式打开文件。

文件重命名

        bool rename(const std::string new_name){return ::rename(_filename.c_str(), new_name.c_str());}static bool rename(const std::string old_name, const std::string new_name){return ::rename(old_name.c_str(), new_name.c_str());}

        使用库中的rename函数即可，一定要前加两个冒号，声明作用域。

文件创建/删除

        bool createFile(){return createFile(_filename);}static bool createFile(const std::string &filename){// 打开文件并立即关闭，创建空文件std::ofstream ofs(filename, std::ios::out);if (!ofs.is_open()){ERROR("%s:文件创建失败", filename.c_str());return false;}ofs.close();return true;}bool removeFile(){return removeFile(_filename);}static bool removeFile(const std::string &filename){return (::remove(filename.c_str()) == 0);}
  1. 文件创建：ofstrea的out打开模式：如果文件不存在就会创建。
  2. 文件删除，调用库中的remove接口即可。

 父级目录的获取

        std::string parentDirectory(){return parentDirectory(_filename);}static std::string parentDirectory(const std::string &filename){size_t pos = filename.find_last_of("/\\");if (pos == std::string::npos)return ".";std::string path = filename.substr(0, pos);return path;}

        寻找最后一个文件目录分割符即可，找不到就返回当前目录。

目录创建/删除

        bool createDirectory(){return createDirectory(parentDirectory(_filename));}static bool createDirectory(const std::string &path){size_t start = 0, end = 0;while (start < path.size()){end = path.find_first_of("/\\", start);if (end == std::string::npos){
#ifdef _WIN32int ret = _mkdir(path.c_str());
#elseint ret = mkdir(path.c_str(), 0775);
#endifif (ret != 0 && errno != EEXIST){ERROR("%s:目录创建失败，错误码: %d", path.c_str(), errno);return false;}break;}if (end == start){start += 1;continue;}std::string subpath = path.substr(0, end);if (exists(path)){start = end + 1;continue;}if (!subpath.empty()){
#ifdef _WIN32int ret = _mkdir(subpath.c_str());
#elseint ret = mkdir(subpath.c_str(), 0775);
#endifif (ret != 0 && errno != EEXIST){ERROR("%s:目录创建失败，错误码: %d", subpath.c_str(), errno);return false;}}start = end + 1;}return true;}bool removeDirectory(){return removeDirectory(parentDirectory(_filename));}static bool removeDirectory(const std::string &path){std::string cmd = "rm -rf " + path;return (system(cmd.c_str()) != -1);}
  1. 目录的创建：需要先创建父级目录才可以创建当前目录，注意如果遇到连续的文件目录分割符需要跳过。
  2. 目录的删除：调用系统命令即可。

附录（完整代码）

#pragma once#include <sqlite3.h>
#include <iostream>
#include <string>
#include <vector>
#include <random>
#include <sstream>
#include <fstream>
#include <atomic>
#include <sys/stat.h>
#include <iomanip>#ifdef _WIN32#include <direct.h>
#endif#include "log.hpp"namespace jiuqi
{class SqliteHelper{public:typedef int (*sqliteCallback)(void *, int, char **, char **);SqliteHelper(const std::string &dbfile): _dbfile(dbfile), _handler(nullptr) {}bool open(int safe_level = SQLITE_OPEN_FULLMUTEX){int ret = sqlite3_open_v2(_dbfile.c_str(), &_handler, SQLITE_OPEN_READWRITE | SQLITE_OPEN_CREATE | safe_level, nullptr);if (ret != SQLITE_OK){// std::cerr << "打开/创建数据库失败: " << sqlite3_errmsg(_handler) << std::endl;ERROR("%s:%s", "打开/创建数据库失败", sqlite3_errmsg(_handler));return false;}return true;}bool exec(const std::string &sql, sqliteCallback callback, void *arg){int ret = sqlite3_exec(_handler, sql.c_str(), callback, arg, nullptr);if (ret != SQLITE_OK){// std::cerr << sql << std::endl;// std::cerr << "执行语句失败: " << sqlite3_errmsg(_handler) << std::endl;ERROR("%s\n%s:%s", sql.c_str(), "执行语句失败: ", sqlite3_errmsg(_handler));return false;}return true;}void close(){if (_handler)sqlite3_close_v2(_handler);}private:std::string _dbfile;sqlite3 *_handler;};class StrHelper{public:static size_t split(const std::string &str, const std::string &sep, std::vector<std::string> &result){size_t end = 0, start = 0;while (start < str.size()){end = str.find(sep, start);if (end == std::string::npos){result.push_back(str.substr(start));break;}if (end != start){result.push_back(str.substr(start, end - start));}start = end + sep.size();}return result.size();}};class UUIDHelper{public:static std::string uuid(){std::random_device rd;std::mt19937_64 gernator(rd());std::uniform_int_distribution<int> distribution(0, 255);std::stringstream ss;for (int i = 1; i <= 8; i++){ss << std::setw(2) << std::setfill('0') << std::hex << distribution(gernator);if (i % 2 == 0)ss << '-';}static std::atomic<size_t> sep(1);size_t num = sep.fetch_add(1);for (int i = 7; i >= 0; i--){ss << std::setw(2) << std::setfill('0') << std::hex << ((num >> (i * 8)) & 0xff);if (i == 4)ss << '-';}return ss.str();}};class FileHelper{public:FileHelper(const std::string &filename) : _filename(filename) {}bool exists(){struct stat st;return (stat(_filename.c_str(), &st) == 0);}static bool exists(const std::string &filename){struct stat st;return (stat(filename.c_str(), &st) == 0);}size_t size(){struct stat st;int ret = stat(_filename.c_str(), &st);if (ret < 0)return 0;return st.st_size;}static size_t size(const std::string &filename){struct stat st;int ret = stat(filename.c_str(), &st);if (ret < 0)return 0;return st.st_size;}bool read(std::string &body){size_t fsize = this->size();return read(body, 0, fsize);}bool read(std::string &body, size_t offset, size_t len){// 打开文件std::ifstream ifs(_filename, std::ios::binary | std::ios::in);if (!ifs.is_open()){ERROR("%s:文件打开失败", _filename.c_str());return false;}// 移动文件读取位置ifs.seekg(offset, std::ios::beg);// 读取文件body.resize(len);ifs.read(&body[0], len);if (!ifs.good()){ERROR("%s:文件读取失败", _filename.c_str());ifs.close();return false;}// 关闭文件返回ifs.close();return true;}static bool read(const std::string &filename, std::string &body){size_t fsize = size(filename);return read(filename, body, 0, fsize);}static bool read(const std::string &filename, std::string &body, size_t offset, size_t len){// 打开文件std::ifstream ifs(filename, std::ios::binary | std::ios::in);if (!ifs.is_open()){ERROR("%s:文件打开失败", filename.c_str());return false;}// 移动文件读取位置ifs.seekg(offset, std::ios::beg);// 读取文件body.resize(len);ifs.read(&body[0], len);if (!ifs.good()){ERROR("%s:文件读取失败", filename.c_str());ifs.close();return false;}// 关闭文件返回ifs.close();return true;}bool write(const std::string &body){size_t fsize = this->size();return write(body.c_str(), fsize, body.size());}bool write(const char *body, size_t offset, size_t len){// 打开文件std::fstream fs(_filename, std::ios::binary | std::ios::out | std::ios::in);if (!fs.is_open()){ERROR("%s:文件打开失败", _filename.c_str());return false;}// 移动文件写入位置fs.seekp(offset, std::ios::beg);// 写入文件fs.write(body, len);if (!fs.good()){ERROR("%s:文件读取失败", _filename.c_str());fs.close();return false;}fs.close();return true;}static bool write(const std::string &filename, const std::string &body){size_t fsize = size(filename);return write(filename, body.c_str(), fsize, body.size());}static bool write(const std::string &filename, const char *body, size_t offset, size_t len){// 打开文件std::fstream fs(filename, std::ios::binary | std::ios::out | std::ios::in);if (!fs.is_open()){ERROR("%s:文件打开失败", filename.c_str());return false;}// 移动文件写入位置fs.seekp(offset, std::ios::beg);// 写入文件fs.write(body, len);if (!fs.good()){ERROR("%s:文件读取失败", filename.c_str());fs.close();return false;}fs.close();return true;}bool rename(const std::string new_name){return ::rename(_filename.c_str(), new_name.c_str());}static bool rename(const std::string old_name, const std::string new_name){return ::rename(old_name.c_str(), new_name.c_str());}bool createFile(){return createFile(_filename);}static bool createFile(const std::string &filename){// 打开文件并立即关闭，创建空文件std::ofstream ofs(filename, std::ios::out);if (!ofs.is_open()){ERROR("%s:文件创建失败", filename.c_str());return false;}ofs.close();return true;}bool removeFile(){return removeFile(_filename);}static bool removeFile(const std::string &filename){return (::remove(filename.c_str()) == 0);}bool createDirectory(){return createDirectory(parentDirectory(_filename));}static bool createDirectory(const std::string &path){size_t start = 0, end = 0;while (start < path.size()){end = path.find_first_of("/\\", start);if (end == std::string::npos){
#ifdef _WIN32int ret = _mkdir(path.c_str());
#elseint ret = mkdir(path.c_str(), 0775);
#endifif (ret != 0 && errno != EEXIST){ERROR("%s:目录创建失败，错误码: %d", path.c_str(), errno);return false;}break;}if (end == start){start += 1;continue;}std::string subpath = path.substr(0, end);if (exists(path)){start = end + 1;continue;}if (!subpath.empty()){
#ifdef _WIN32int ret = _mkdir(subpath.c_str());
#elseint ret = mkdir(subpath.c_str(), 0775);
#endifif (ret != 0 && errno != EEXIST){ERROR("%s:目录创建失败，错误码: %d", subpath.c_str(), errno);return false;}}start = end + 1;}return true;}bool removeDirectory(){return removeDirectory(parentDirectory(_filename));}static bool removeDirectory(const std::string &path){std::string cmd = "rm -rf " + path;return (system(cmd.c_str()) != -1);}std::string parentDirectory(){return parentDirectory(_filename);}static std::string parentDirectory(const std::string &filename){size_t pos = filename.find_last_of("/\\");if (pos == std::string::npos)return ".";std::string path = filename.substr(0, pos);return path;}private:std::string _filename;};
}

http://www.xdnf.cn/news/15829.html

相关文章：

  • 服务器上的文件复制到本地 Windows 系统
  • python网络爬虫小项目（爬取评论）超级简单
  • git fork的项目远端标准协作流程 仓库设置[设置成upstream]
  • 快速上手git
  • LINUX入门（二）QT的安装及运行环境搭建
  • 【实习总结】Qt中如何使用QSettings操作.ini配置文件
  • Vue中组件的生命周期
  • 08_Opencv_基本图形绘制
  • Docker实战：使用Docker部署envlinks极简个人导航页
  • 激光雷达和相机在线标定
  • [C/C++安全编程]_[中级]_[如何安全使用循环语句]
  • 语言学校为何成为IT润日路径的制度跳板？签证与迁移结构的工程化拆解
  • 交通出行大前端与 AI 融合：智能导航与出行预测
  • 智能制造——48页毕马威：汽车营销与研发数字化研究【附全文阅读】
  • jxORM--编程指南
  • linux + 宝塔面板 部署 django网站 启动方式:uwsgi 和gunicorn如何选择 ?
  • windows命令提示符cmd使用
  • Django接口自动化平台实现（四）
  • 第 30 场 蓝桥·算法入门赛 题解
  • 制作mac 系统U盘
  • 零基础学习性能测试第一章-为什么会有性能问题
  • 全面解析 JDK 提供的 JVM 诊断与故障处理工具
  • VSCode使用Jupyter完整指南配置机器学习环境
  • 秒赤Haproxy配置算法
  • `TransportService` 是 **Elasticsearch 传输层的“中枢路由器”**
  • SparseTSF：用 1000 个参数进行长序列预测建模
  • RabbitMQ面试精讲 Day 4：Queue属性与消息特性
  • Java拓扑排序：2115 从给定原材料中找到所有可以做出的菜
  • LWJGL教程（2）——游戏循环
  • 网络（HTTP）