message.hpp

浏览该文件的文档.

 
#pragma once
#include "../common/logger.hpp"
#include "../common/helper.hpp"
#include "../common/msg.pb.h"
#include <iostream>
#include <unordered_map>
#include <mutex>
#include <memory>
#include <list>
 
namespace XuMQ
{
    const char *DATAFILE_SUBFIX = ".mqd";              
    const char *TMPFILE_SUBFIX = ".mqd.tmp";           
    const char *MSG_VALID = "1";                       
    const char *MSG_INVALID = "0";                     
    using MessagePtr = std::shared_ptr<XuMQ::Message>; 
    class MessageMapper
    {
    public:
        MessageMapper(std::string &basedir, const std::string &qname)
            : _qname(qname)
        {
            if (basedir.back() != '/' && basedir.back() != '\\')
                basedir.push_back('/');
            _datafile = basedir + qname + DATAFILE_SUBFIX;
            _tmpfile = basedir + qname + TMPFILE_SUBFIX;
 
            if (FileHelper(basedir).exists() == false)
            {
                int ret = FileHelper::createDirectory(basedir);
                if (ret == false)
                {
                    fatal(logger, "创建文件夹失败!");
                    abort();
                }
            }
            createMsgFile();
        }
        bool createMsgFile()
        {
            if (FileHelper(_datafile).exists() == false)
            {
                bool ret = FileHelper::createFile(_datafile);
                if (ret == false)
                {
                    error(logger, " %s :创建队列数据文件失败!", _datafile.c_str());
                    return false;
                }
            }
            return true;
        }
        void removeMsgFile()
        {
            FileHelper::removeFile(_datafile);
            FileHelper::removeFile(_tmpfile);
        }
        bool insert(const MessagePtr &msg)
        {
            return insert(_datafile, msg);
        }
        bool remove(MessagePtr &msg)
        {
            // 将msg中的有效标志为设置为'0'(false)
            msg->mutable_payload()->set_valid(MSG_INVALID);
            // 对msg进行序列化
            std::string body = msg->payload().SerializeAsString();
            if (body.size() != msg->length())
            {
                error(logger, "不能修改文件中的数据信息, 新生成的数据与原数据长度不一致!");
                return false;
            }
            // 将序列化的消息 写入到数据中的指定位置(覆盖原有的数据)
            FileHelper helper(_datafile);
            bool ret = helper.write(body.c_str(), msg->offset(), body.size());
            if (ret == false)
            {
                error(logger, " %s :队列数据文件写入失败!", _datafile.c_str());
                return false;
            }
            return true;
        }
        std::list<MessagePtr> garbageCollection()
        {
            std::list<MessagePtr> result;
            // 加载文件中所有的有效数据 存储格式 4字节长度|数据|4字节长度|数据...
            bool ret = load(result);
            if (ret == false)
            {
                error(logger, "加载有效数据失败!");
                return result;
            }
            // 有效数据进行序列化存储到临时文件中
            FileHelper::createFile(_tmpfile);
            for (auto &msg : result)
            {
                ret = insert(_tmpfile, msg);
                if (ret == false)
                {
                    error(logger, " %s :临时文件写入消息数据失败!", _tmpfile);
                    return result;
                }
            }
            // 删除原文件
            ret = FileHelper::removeFile(_datafile);
            if (ret == false)
            {
                error(logger, " %s :删除原文件失败!", _datafile);
                return result;
            }
 
            // 修改临时文件名为原文件名称
            ret = FileHelper(_tmpfile).rename(_datafile);
            if (ret == false)
            {
                error(logger, " %s :修改临时文件名称失败!", _tmpfile.c_str());
                return result;
            }
            // 返回新的有效数据
            return result;
        }
 
    private:
        bool insert(const std::string &filename, const MessagePtr &msg)
        {
            // 新增数据添加在文件末尾
            // 消息序列化
            std::string body = msg->payload().SerializeAsString();
            // 获取文件长度
            FileHelper helper(filename);
            size_t fsize = helper.size();
            // 先写入4字节数据长度
            size_t message_size = body.size();
            bool ret = helper.write((char *)&message_size, fsize, sizeof(size_t));
            if (ret == false)
            {
                error(logger, " %s :队列数据文件写入长度失败!", filename.c_str());
                return false;
            }
 
            // 写入数据到指定位置
            ret = helper.write(body.c_str(), fsize + sizeof(size_t), body.size());
            if (ret == false)
            {
                error(logger, " %s :队列数据文件写入内容失败!", filename.c_str());
                return false;
            }
            // 更新msg中的存储信息
            msg->set_offset(fsize + sizeof(size_t));
            msg->set_length(body.size());
            return true;
        }
        bool load(std::list<MessagePtr> &result)
        {
            FileHelper helper(_datafile);
            size_t offset = 0, msg_size;
            size_t fsize = helper.size();
            bool ret;
            while (offset < fsize)
            {
                ret = helper.read((char *)&msg_size, offset, sizeof(size_t));
                if (ret == false)
                {
                    error(logger, " %s :读取消息长度失败!", _datafile);
                    return false;
                }
                offset += sizeof(size_t);
                std::string msg_body(msg_size, '\0');
                ret = helper.read(&msg_body[0], offset, msg_size);
                if (ret == false)
                {
                    error(logger, " %s :读取消息数据失败!", _datafile);
                    return false;
                }
                offset += msg_size;
                MessagePtr msgp = std::make_shared<Message>();
                msgp->mutable_payload()->ParseFromString(msg_body);
                if (msgp->payload().valid() == MSG_INVALID) // 无效消息则处理下一个
                    continue;
                result.push_back(msgp); // 有效消息保存
            }
            return true;
        }
 
    private:
        std::string _qname;    
        std::string _datafile; 
        std::string _tmpfile;  
    };
 
    class QueueMessage
    {
    public:
        using ptr = std::shared_ptr<QueueMessage>;
        QueueMessage(std::string &basedir, const std::string &qname)
            : _mapper(basedir, qname), _qname(qname), _valid_count(0), _total_count(0)
        {
        }
        void recovery()
        {
            std::unique_lock<std::mutex> lock(_mutex);
            _msgs = _mapper.garbageCollection();
            for (auto &msg : _msgs)
                _durable_msgs.insert(std::make_pair(msg->payload().properties().id(), msg));
            _valid_count = _total_count = _msgs.size();
        }
        bool insert(const BasicProperties *bp, const std::string &body, bool queue_id_durable)
        {
            // 构造消息对象
            MessagePtr msg = std::make_shared<Message>();
            msg->mutable_payload()->set_body(body);
            if (bp != nullptr)
            {
                DeliveryMode mode = queue_id_durable ? bp->delivery_mode() : DeliveryMode::UNDURABLE;
                msg->mutable_payload()->mutable_properties()->set_id(bp->id());
                msg->mutable_payload()->mutable_properties()->set_delivery_mode(mode);
                msg->mutable_payload()->mutable_properties()->set_routing_key(bp->routing_key());
            }
            else
            {
                DeliveryMode mode = queue_id_durable ? DeliveryMode::DURABLE : DeliveryMode::UNDURABLE;
                msg->mutable_payload()->mutable_properties()->set_id(UUIDHelper::uuid());
                msg->mutable_payload()->mutable_properties()->set_delivery_mode(mode);
                msg->mutable_payload()->mutable_properties()->set_routing_key("");
            }
            std::unique_lock<std::mutex> lock(_mutex);
            // 判断消息是否需要持久化
            if (msg->payload().properties().delivery_mode() == DeliveryMode::DURABLE)
            {
                msg->mutable_payload()->set_valid(MSG_VALID); // 持久化存储中表示数据有效
                // 持久化存储
                bool ret = _mapper.insert(msg);
                if (ret == false)
                {
                    error(logger, " %s :持久化存储消息失败!", body.c_str());
                    return false;
                }
                _valid_count++;
                _total_count++;
                _durable_msgs.insert(std::make_pair(msg->payload().properties().id(), msg));
            }
            // 内存管理
            _msgs.push_back(msg);
            return true;
        }
        MessagePtr front()
        {
            if (_msgs.size() == 0)
                return MessagePtr();
            std::unique_lock<std::mutex> lock(_mutex);
            // 获取队头消息 从msgs取出数据
            MessagePtr msg = _msgs.front();
            _msgs.pop_front();
            // 将消息对象插入待确认映射表 等到收到确认ack后删除
            _waitack_msgs.insert(std::make_pair(msg->payload().properties().id(), msg));
            return msg;
        }
        bool remove(const std::string &msg_id)
        {
            std::unique_lock<std::mutex> lock(_mutex);
            // 从待确认映射表中查找消息
            auto it = _waitack_msgs.find(msg_id);
            if (it == _waitack_msgs.end())
            {
                warn(logger, "没有找到要删除的消息! 消息id: %s", msg_id.c_str());
                return true;
            }
            // 查看持久化模式
            if (it->second->payload().properties().delivery_mode() == DeliveryMode::DURABLE)
            {
                // 删除持久化信息
                _mapper.remove(it->second);
                _durable_msgs.erase(msg_id);
                _valid_count--;
                garbageCollection();
            }
            // 删除内存中的信息
            _waitack_msgs.erase(msg_id);
            return true;
        }
        size_t availableCount()
        {
            std::unique_lock<std::mutex> lock(_mutex);
            return _msgs.size();
        }
        size_t totalCount()
        {
            std::unique_lock<std::mutex> lock(_mutex);
            return _total_count;
        }
        size_t waitAckCount()
        {
            std::unique_lock<std::mutex> lock(_mutex);
            return _waitack_msgs.size();
        }
        size_t durableCount()
        {
            std::unique_lock<std::mutex> lock(_mutex);
            return _durable_msgs.size();
        }
        void clear()
        {
            std::unique_lock<std::mutex> lock(_mutex);
            _mapper.removeMsgFile();
            _msgs.clear();
            _durable_msgs.clear();
            _waitack_msgs.clear();
            _valid_count = 0;
            _total_count = 0;
        }
 
    private:
        bool garbageCollectionCheck()
        {
            if (_total_count > 2000 && _valid_count * 10 / _total_count < 5)
                return true;
            return false;
        }
        void garbageCollection()
        {
            // 垃圾回收 获取有效消息信息链表
            if (garbageCollectionCheck() == false)
                return;
            // 更新消息的实际存储位置
            std::list<MessagePtr> msgs = _mapper.garbageCollection();
            for (auto &msg : msgs)
            {
                auto it = _durable_msgs.find(msg->payload().properties().id());
                if (it == _durable_msgs.end())
                {
                    _msgs.push_back(msg);
                    _durable_msgs.insert(std::make_pair(msg->payload().properties().id(), msg));
                    info(logger, "垃圾回收后 有一条消息在内存尚未被管理 已插入待推送消息列表");
                    continue;
                }
                it->second->set_offset(msg->offset());
                it->second->set_length(msg->length());
            }
            // 更新有效消息数量 总持久化消息数量
            _valid_count = _total_count = msgs.size();
        }
 
    private:
        std::mutex _mutex;                                         
        std::string _qname;                                        
        size_t _valid_count;                                       
        size_t _total_count;                                       
        MessageMapper _mapper;                                     
        std::list<MessagePtr> _msgs;                               
        std::unordered_map<std::string, MessagePtr> _durable_msgs; 
        std::unordered_map<std::string, MessagePtr> _waitack_msgs; 
    };
 
    class MessageManager
    {
    public:
        using ptr = std::shared_ptr<MessageManager>; 
        MessageManager(const std::string &basedir) : _basedir(basedir) {}
        void initQueueMessage(const std::string &qname)
        {
            QueueMessage::ptr qmp;
            {
                std::unique_lock<std::mutex> lock(_mutex);
                auto it = _queue_msgs.find(qname);
                if (it != _queue_msgs.end())
                    return;
                qmp = std::make_shared<QueueMessage>(_basedir, qname);
                _queue_msgs.insert(std::make_pair(qname, qmp));
            }
            qmp->recovery();
        }
        void destroyQueueMessage(const std::string &qname)
        {
            QueueMessage::ptr qmp;
            {
                std::unique_lock<std::mutex> lock(_mutex);
                auto it = _queue_msgs.find(qname);
                if (it == _queue_msgs.end())
                    return;
                qmp = it->second;
                _queue_msgs.erase(qname);
            }
            qmp->clear();
        }
        bool insert(const std::string &qname, BasicProperties *bp, const std::string &body, bool mode)
        {
            QueueMessage::ptr qmp;
            {
                std::unique_lock<std::mutex> lock(_mutex);
                auto it = _queue_msgs.find(qname);
                if (it == _queue_msgs.end())
                {
                    error(logger, "插入消息失败, 没有找到 %s 队列", qname.c_str());
                    return false;
                }
                qmp = it->second;
            }
            return qmp->insert(bp, body, mode);
        }
        MessagePtr front(const std::string &qname)
        {
            QueueMessage::ptr qmp;
            {
                std::unique_lock<std::mutex> lock(_mutex);
                auto it = _queue_msgs.find(qname);
                if (it == _queue_msgs.end())
                {
                    error(logger, "获取队头消息失败, 没有找到 %s 队列", qname.c_str());
                    return MessagePtr();
                }
                qmp = it->second;
            }
            return qmp->front();
        }
        void ack(const std::string &qname, const std::string &msg_id)
        {
            QueueMessage::ptr qmp;
            {
                std::unique_lock<std::mutex> lock(_mutex);
                auto it = _queue_msgs.find(qname);
                if (it == _queue_msgs.end())
                {
                    error(logger, "确认消息失败, 没有找到 %s 队列", qname.c_str());
                    return;
                }
                qmp = it->second;
            }
            qmp->remove(msg_id);
        }
 
        size_t availableCount(const std::string &qname)
        {
            QueueMessage::ptr qmp;
            {
                std::unique_lock<std::mutex> lock(_mutex);
                auto it = _queue_msgs.find(qname);
                if (it == _queue_msgs.end())
                {
                    error(logger, "获取可获取消息数量失败, 没有找到 %s 队列", qname.c_str());
                    return 0;
                }
                qmp = it->second;
            }
            return qmp->availableCount();
        }
        size_t totalCount(const std::string &qname)
        {
            QueueMessage::ptr qmp;
            {
                std::unique_lock<std::mutex> lock(_mutex);
                auto it = _queue_msgs.find(qname);
                if (it == _queue_msgs.end())
                {
                    error(logger, "获取总消息数量失败, 没有找到 %s 队列", qname.c_str());
                    return 0;
                }
                qmp = it->second;
            }
            return qmp->totalCount();
        }
        size_t waitAckCount(const std::string &qname)
        {
            QueueMessage::ptr qmp;
            {
                std::unique_lock<std::mutex> lock(_mutex);
                auto it = _queue_msgs.find(qname);
                if (it == _queue_msgs.end())
                {
                    error(logger, "获取待确认消息数量失败, 没有找到 %s 队列", qname.c_str());
                    return 0;
                }
                qmp = it->second;
            }
            return qmp->waitAckCount();
        }
        size_t durableCount(const std::string &qname)
        {
            QueueMessage::ptr qmp;
            {
                std::unique_lock<std::mutex> lock(_mutex);
                auto it = _queue_msgs.find(qname);
                if (it == _queue_msgs.end())
                {
                    error(logger, "获取持久化消息数量失败, 没有找到 %s 队列", qname.c_str());
                    return 0;
                }
                qmp = it->second;
            }
            return qmp->durableCount();
        }
        void clear()
        {
            QueueMessage::ptr qmp;
            for (auto &qmsg : _queue_msgs)
            {
                qmsg.second->clear();
            }
        }
 
    private:
        std::mutex _mutex;                                              
        std::string _basedir;                                           
        std::unordered_map<std::string, QueueMessage::ptr> _queue_msgs; 
    };
}