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

目录

日志打印工具

实用 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 _WIN32
                    int ret = _mkdir(path.c_str());
#else
                    int ret = mkdir(path.c_str(), 0775);
#endif

                    if (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 _WIN32
                    int ret = _mkdir(subpath.c_str());
#else
                    int ret = mkdir(subpath.c_str(), 0775);
#endif

                    if (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 _WIN32
                    int ret = _mkdir(path.c_str());
#else
                    int ret = mkdir(path.c_str(), 0775);
#endif

                    if (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 _WIN32
                    int ret = _mkdir(subpath.c_str());
#else
                    int ret = mkdir(subpath.c_str(), 0775);
#endif

                    if (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;
    };
}