Asio14-ImplementLogicSystem.md
单例逻辑类
因为我们的逻辑层一般是单线程(如果多线程,那么在处理资源的情况下,需要对共享资源加锁,更复杂),因此我们实现一个单例模板类,后期可能有其他也需要单例.
首先是一个单例的模板类.
#pragma once#include <iostream>
#include <memory>
#include <mutex>template <typename T>
class Singleton {
private:Singleton(const Singleton<T>&) = delete;Singleton<T>& operator=(const Singleton<T>&) = delete;Singleton(Singleton<T>&&) = delete;Singleton<T>& operator=(Singleton<T>&&) = delete;protected:Singleton() = default;~Singleton() = default;private:inline static std::shared_ptr<T> _instance = nullptr;public:static std::shared_ptr<T> GetInstance(){static std::once_flag flag;std::call_once(flag, [&]() {_instance = std::make_shared<T>();});return _instance;}
};那么实现逻辑类的具体单例类LogicSystem,我们继承自单例模板类.
// .h
typedef std::function<void(std::shared_ptr<CSession>, short msg_id, const std::string& msg)> FuncBack;class LogicSystem : public Singleton<LogicSystem> {// friend class Singleton<LogicSystem>;template <typename T>friend class Singleton;public:void PostMsgToQue(std::shared_ptr<LogicNode> msg);void RegisterCallBacks();void HelloWorldCallBacks(std::shared_ptr<CSession>, short msg_id, const std::string& msg);void DealMsg();public:LogicSystem();~LogicSystem();private:std::queue<std::shared_ptr<LogicNode>> _queue;std::mutex _mutex;std::condition_variable _cv;std::thread _worker_thread;bool _stop;std::unordered_map<short, FuncBack> _function_callbacks;
};LogicNode定义在CSession.h中
class LogicNode {friend class LogicSystem;
public:LogicNode(shared_ptr<CSession>, shared_ptr<RecvNode>);
private:shared_ptr<CSession> _session;shared_ptr<RecvNode> _recvnode;
};
LogicSystem构造函数如下:
LogicSystem::LogicSystem(): _stop(false)
{RegisterCallBacks();_worker_thread = std::thread(&LogicSystem::DealMsg, this);
}
先注册消息处理函数,然后启动工作线程,工作线程的执行操作为DealMsg.
LogicSystem::RegisterCallBacks函数如下:
void LogicSystem::RegisterCallBacks()
{_function_callbacks[MSG_HELLO_WORD] = std::bind(&LogicSystem::HelloWorldCallBacks, this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3);
}
当前仅仅注册了一个id为1001的处理函数.
LogicSystem::HelloWordCallBack函数如下:
void LogicSystem::HelloWorldCallBacks(std::shared_ptr<CSession> session, short msg_id, const std::string& msg)
{Json::Reader reader;Json::Value root;reader.parse(msg, root);std::cout << "Received msg:" << root["id"] << "\t" << root["msg"].asCString() << std::endl;std::cout << "Raw msg : \n"<< msg << std::endl;std::string originalMsg = root["msg"].asString();std::string newMsg = root.toStyledString();session->Send(originalMsg, root["id"].asInt());
}
这个回调是解析消息,并且将消息处理后,发送给客户端.
LogicSystem::DealMsg(工作线程的执行函数)如下:
void LogicSystem::DealMsg()
{while (true) {std::unique_lock<std::mutex> lock(_mutex);_cv.wait(lock, [this] { return _stop || !_queue.empty(); });if (_stop && _queue.empty()) {break; // 停止且队列为空时才退出}if (!_queue.empty()) {auto msg = _queue.front();_queue.pop();lock.unlock(); // 尽早释放锁// 统一处理逻辑auto it = _function_callbacks.find(msg->_recv->_msg_id);if (it != _function_callbacks.end()) {it->second(msg->_session, msg->_recv->_msg_id,std::string(msg->_recv->_data, msg->_recv->_total_len));}}}
}
这段代码的逻辑是,从消息处理队列中取出消息节点进行处理.每次循环进入执行操作时,有两种情况,一是逻辑层要结束,_stop=true;二是消息处理队列不为空,需要处理.
对于上面队列非空,if分区的操作而言,假设有多个待处理,我们需要多次循环,多次等待锁,获得锁,处理,解锁.为了避免频繁的上锁解锁,我们可以采用局部队列.通过将此时消息队列的待处理节点和局部队列swap,那么此时就可以直接解除锁,同时工作线程统一循环处理,减少了等待锁的时间,效率更高.
/** 这段代码相比上面代码而言,使用局部队列与消息队列swap,这时候就可以释放锁,大大地较少等待锁的时间 */
if (!_queue.empty()) {std::queue<std::shared_ptr<LogicNode>> local_queue;local_queue.swap(_queue);lock.unlock(); // 尽早释放锁while (!local_queue.empty()) {auto msg = local_queue.front();local_queue.pop();// 统一处理逻辑auto it = _function_callbacks.find(msg->_recv->_msg_id);if (it != _function_callbacks.end()) {it->second(msg->_session, msg->_recv->_msg_id,std::string(msg->_recv->_data, msg->_recv->_total_len));}}
}