consumer.hpp

浏览该文件的文档.

 
 
#pragma once
#include "../common/logger.hpp"
#include "../common/helper.hpp"
#include "../common/msg.pb.h"    
#include <iostream>
#include <unordered_map>
#include <vector>
#include <mutex> 
#include <memory>               
#include <functional>  
 
namespace XuMQ
{
    using ConsumerCallback = std::function<void(const std::string&, const BasicProperties *, const std::string&)>; 
    struct Consumer
    {
        using ptr = std::shared_ptr<Consumer>; 
        std::string tag;                       
        std::string qname;                     
        bool auto_ack;                         
        ConsumerCallback callback;             
 
        Consumer() {}
        Consumer(const std::string &ctag, const std::string &queue_name, bool ack, const ConsumerCallback &cb)
            : tag(ctag), qname(queue_name), auto_ack(ack), callback(cb) {}
    };
 
    class QueueConsumer
    {
    public:
        using ptr = std::shared_ptr<QueueConsumer>; 
        QueueConsumer(const std::string &qname)
            : _qname(qname), _rr_seq(0) {}
        Consumer::ptr create(const std::string &ctag, const std::string &queue_name, bool ack, const ConsumerCallback &cb)
        {
            // 加锁
            std::unique_lock<std::mutex> lock(_mutex);
            // 判断消费者是否重复
            for (auto &consumer : _consumers)
            {
                if (consumer->tag == ctag)
                {
                    warn(logger, "消费者重复添加! 消费者标识: %s", ctag.c_str());
                    return Consumer::ptr();
                }
            }
            // 构造对象
            auto consumer = std::make_shared<Consumer>(ctag, queue_name, ack, cb);
            // 添加消费者
            _consumers.push_back(consumer);
            return consumer;
        }
        void remove(const std::string &ctag)
        {
            std::unique_lock<std::mutex> lock(_mutex);
            for (auto it = _consumers.begin(); it != _consumers.end(); it++)
            {
                if (ctag == (*it)->tag)
                {
                    _consumers.erase(it);
                    return;
                }
            }
        }
        Consumer::ptr choose()
        {
            // 加锁
            std::unique_lock<std::mutex> lock(_mutex);
            if (_consumers.size() == 0)
            {
                error(logger, "当前消费者队列为空!");
                return Consumer::ptr();
            }
            // 获取轮转到的下标
            int idx = _rr_seq++ % _consumers.size();
            // 返回
            return _consumers[idx];
        }
        bool empty()
        {
            // 加锁
            std::unique_lock<std::mutex> lock(_mutex);
            return _consumers.size() == 0;
        }
        bool exists(const std::string &ctag)
        {
            // 加锁
            std::unique_lock<std::mutex> lock(_mutex);
            for (auto &consumer : _consumers)
                if (consumer->tag == ctag)
                    return true;
            return false;
        }
        void clear()
        {
            // 加锁
            std::unique_lock<std::mutex> lock(_mutex);
            _consumers.clear();
            _rr_seq = 0;
        }
 
    private:
        std::string _qname;                    
        std::mutex _mutex;                     
        uint64_t _rr_seq;                      
        std::vector<Consumer::ptr> _consumers; 
    };
 
    class ConsumerManager
    {
    public:
        using ptr = std::shared_ptr<ConsumerManager>; 
        ConsumerManager() {}
        void initQueueConsumer(const std::string &name)
        {
            // 加锁
            std::unique_lock<std::mutex> lock(_mutex);
            // 判断重复
            auto it = _qconsumers.find(name);
            if (it != _qconsumers.end())
            {
                return;
            }
            // 新增
            auto qconsumers = std::make_shared<QueueConsumer>(name);
            _qconsumers.insert(std::make_pair(name, qconsumers));
        }
        void destroyQueueConsumer(const std::string &name)
        {
            std::unique_lock<std::mutex> lock(_mutex);
            _qconsumers.erase(name);
        }
        Consumer::ptr create(const std::string &ctag, const std::string &queue_name, bool ack, const ConsumerCallback &cb)
        {
            QueueConsumer::ptr qcp;
            {
                std::unique_lock<std::mutex> lock(_mutex);
                // 获取队列的消费者管理单元
                auto it = _qconsumers.find(queue_name);
                if (it == _qconsumers.end())
                {
                    warn(logger, "没有找到指定队列! 队列名称: %s", queue_name.c_str());
                    return Consumer::ptr();
                }
                qcp = it->second;
            }
            // 完成新建
            return qcp->create(ctag, queue_name, ack, cb);
        }
        void remove(const std::string &ctag, const std::string &queue_name)
        {
            QueueConsumer::ptr qcp;
            {
                std::unique_lock<std::mutex> lock(_mutex);
                // 获取队列的消费者管理单元
                auto it = _qconsumers.find(queue_name);
                if (it == _qconsumers.end())
                {
                    warn(logger, "没有找到指定队列! 队列名称: %s", queue_name.c_str());
                    return;
                }
                qcp = it->second;
            }
            qcp->remove(ctag);
        }
        Consumer::ptr choose(const std::string &queue_name)
        {
            QueueConsumer::ptr qcp;
            {
                std::unique_lock<std::mutex> lock(_mutex);
                // 获取队列的消费者管理单元
                auto it = _qconsumers.find(queue_name);
                if (it == _qconsumers.end())
                {
                    warn(logger, "没有找到指定队列! 队列名称: %s", queue_name.c_str());
                    return Consumer::ptr();
                }
                qcp = it->second;
            }
            return qcp->choose();
        }
 
        bool empty(const std::string &queue_name)
        {
            QueueConsumer::ptr qcp;
            {
                std::unique_lock<std::mutex> lock(_mutex);
                // 获取队列的消费者管理单元
                auto it = _qconsumers.find(queue_name);
                if (it == _qconsumers.end())
                {
                    warn(logger, "没有找到指定队列! 队列名称: %s", queue_name.c_str());
                    return true;
                }
                qcp = it->second;
            }
            return qcp->empty();
        }
        bool exists(const std::string &ctag, const std::string &queue_name)
        {
            QueueConsumer::ptr qcp;
            {
                std::unique_lock<std::mutex> lock(_mutex);
                // 获取队列的消费者管理单元
                auto it = _qconsumers.find(queue_name);
                if (it == _qconsumers.end())
                {
                    warn(logger, "没有找到指定队列! 队列名称: %s", queue_name.c_str());
                    return false;
                }
                qcp = it->second;
            }
            return qcp->exists(ctag);
        }
        void clear()
        {
            std::unique_lock<std::mutex> lock(_mutex);
            QueueConsumer::ptr qcp;
            _qconsumers.clear();
        }
 
    private:
        std::mutex _mutex;
        std::unordered_map<std::string, QueueConsumer::ptr> _qconsumers;
    };
}