1、增加线程池，用于处理消息

CMakeLists.txt

@@ -22,6 +22,7 @@ add_executable(dcapp
         ${PROJECT_SOURCE_DIR}/datacenter/src/queue.cpp
         ${PROJECT_SOURCE_DIR}/datacenter/src/mqttClient.cpp
         ${PROJECT_SOURCE_DIR}/datacenter/src/ToolUtils.cpp
+		${PROJECT_SOURCE_DIR}/datacenter/src/thread_pool.cpp
         ${PROJECT_SOURCE_DIR}/datacenter/bizService/F_InsertRecord.cpp
 	${PROJECT_SOURCE_DIR}/datacenter/bizService/F_UpdateRecord.cpp
         ${PROJECT_SOURCE_DIR}/datacenter/bizService/F_GetSaveCount.cpp

datacenter/inc/datacenter.h

@@ -11,13 +11,15 @@
 #define _DCAPP_DATACENTER_DATACENTER_H_
 
 #include <unistd.h>
+#include <thread>
+#include <functional>
 #include "service.h"
 #include "mqttClient.h"
 #include "queue.h"
+#include "thread_pool.h"
 
 #define POOLSIZE 10
 
-
 class DataCenter : public IService
 {
 public:
@@ -45,12 +47,11 @@ public:
     virtual void Exec()
     {
         datacenterinit();
-        while (1)
-        {
-            ExecPoll();
-            ExecReponse();
-            usleep(500000);
-        }
+
+        std::thread recvThread(&DataCenter::ExecPoll,this);
+        std::thread sendThread(&DataCenter::ExecReponse, this);
+        recvThread.join();
+        sendThread.join();
     }
 
     virtual const char *getServiceName()
@@ -58,14 +59,20 @@ public:
         return "DataCenter";
     }
 
+public:
+    static void ExecPoll(void *);       // 执行轮巡业务
+    static void ExecReponse(void*);    // 执行应答任务
+//    static bool handleMsg(msg_mqtt_t msg);
 private:
     void datacenterinit(void); // 初始化
-    void ExecPoll(void);       // 执行轮巡业务
-    void ExecReponse(void);    // 执行应答任务
-    bool handleMsg(msg_mqtt_t msg);
+
+private:
+    bool recvThreadFlag = false;
+    bool sendThreadFlag = false;
     MessageQueue mQueue_In;    // 输入队列
     MessageQueue mQueue_Out;  // 输出队列
     CMqttClient *objmqttClient;
+    ThreadPool *objThreadPool;
     
 };
 

datacenter/inc/thread_pool.h

@@ -0,0 +1,73 @@
+#ifndef THREAD_POOL_H
+#define THREAD_POOL_H
+
+#include <vector>
+#include <queue>
+#include <memory>
+#include <thread>
+#include <mutex>
+#include <condition_variable>
+#include <future>
+#include <functional>
+#include <stdexcept>
+#include <atomic>
+#include <chrono>
+#include <iostream>
+
+#define MAX_THREAD_POOL_SIZE 5
+class ThreadPool {
+public:
+    ThreadPool();
+    ThreadPool(const ThreadPool&) = delete;
+    ThreadPool& operator=(const ThreadPool&) = delete;
+
+    ~ThreadPool();
+
+
+    void setTaskQueMaxThreshold(int Threshold){
+        // 设置最大任务数量
+        taskQueMaxThreshold_ = Threshold;
+    }
+
+    void start(int initThreadNum=std::thread::hardware_concurrency());
+
+    template <typename Func, typename... Args>
+    auto submitTask(Func&& func, Args&&... args) -> std::future<decltype(func(args...))>{
+        using returnType = decltype(func(args...));
+        auto task_ptr = std::make_shared<std::packaged_task<returnType()>>(std::bind(std::forward<Func>(func), std::forward<Args>(args)...));
+        std::future<returnType> result = task_ptr -> get_future();
+
+        std::unique_lock<std::mutex> ulock(taskQueMtx_);
+        if (taskNum_ >= taskQueMaxThreshold_) {
+            std::cerr << "[Thread Pool] Task queue is full, task submitting failed!" << " queue size: " << taskQue_.size() <<
+                      ", taskQueMaxThreshold_: " << taskQueMaxThreshold_ <<  std::endl;
+            return std::future<returnType>();
+        }
+
+        taskQue_.emplace(std::make_shared<std::function<void()>>([task_ptr](){
+            (*task_ptr)();
+        }));
+        ++ taskNum_;
+        notEmpty_.notify_one();
+        return result;
+    }
+
+    void threadFunc();
+
+private:
+    std::vector<std::shared_ptr<std::thread>> threads_; // 线程列表
+    size_t initThreadNum_; //初始线程数量
+    using Task = std::function<void()>;
+    std::queue<std::shared_ptr<Task>> taskQue_; // 任务队列
+    std::atomic_int taskNum_; // 任务数量
+    int taskQueMaxThreshold_; // 最大任务数量
+
+    std::mutex taskQueMtx_; // 任务队列互斥锁
+    std::condition_variable notFull_;
+    std::condition_variable notEmpty_;
+
+    std::atomic_bool stop_; // 用于关闭线程池
+
+};
+
+#endif // THREAD_POOL_H 

datacenter/src/datacenter.cpp

@@ -9,6 +9,7 @@
  */
 #include <iostream>
 #include <unistd.h>
+#include <functional>
 
 #include "datacenter.h"
 #include "msg.h"
@@ -30,42 +31,19 @@ int WriteResponseMsg(msg_t msg) {
 void DataCenter::datacenterinit(void) {
     this->objmqttClient = new CMqttClient(&mQueue_In);
     if (NULL == objmqttClient)
-        return;
+        exit(-1);
+
+    this->objThreadPool = new ThreadPool();
+    if (NULL == objThreadPool){
+        delete objmqttClient;
+        exit(-1);
+    }
+    this->objThreadPool->setTaskQueMaxThreshold(MAX_THREAD_POOL_SIZE);
+    objThreadPool->start();
 
     objmqttClient->connectServer();
     //todo 发送上线通知
 }
-
-void DataCenter::ExecPoll(void) {
-    if(!this->objmqttClient->getConnectStatus()){
-        return;
-    }
-    msg_mqtt_t inMsg = this->mQueue_In.pop_nonblock();
-    if(inMsg.topic.empty() && inMsg.payload.empty()){
-//        std::cout << "mQueue_In:" << " : " << " is empty " << std::endl;
-        return;
-    }
-
-    this->handleMsg(inMsg);
-}
-
-void DataCenter::ExecReponse() {
-
-    if(!this->objmqttClient->getConnectStatus()){
-        return;
-    }
-
-    //消息应答队列无阻赛返回
-    msg_mqtt_t outMsg = this->mQueue_Out.pop_nonblock();
-    if(outMsg.topic.empty() && outMsg.payload.empty()){
-//        std::cout << "mQueue_Out:" << " : " << " is empty " << std::endl;
-        return;
-    }
-
-    std::cout << "send:" << outMsg.topic << " : " << outMsg.payload << std::endl;
-    this->objmqttClient->publish(outMsg.topic,outMsg.payload);
-}
-
 std::string getPidFromTopic(std::string topic){
     size_t pos = topic.find_last_of("/");
     if(pos != std::string::npos){
@@ -73,7 +51,7 @@ std::string getPidFromTopic(std::string topic){
     }
     return "";
 }
-bool DataCenter::handleMsg(msg_mqtt_t inMqttMsg) {
+bool handleMsg(msg_mqtt_t inMqttMsg,MessageQueue &mQueue_Out) {
     int fRet = -1;
     msg_t intMsg,outMsg;
 
@@ -104,6 +82,31 @@ bool DataCenter::handleMsg(msg_mqtt_t inMqttMsg) {
     msg_mqtt_t outMqttMsg;
     outMqttMsg.topic = outMsg.topic;
     outMqttMsg.payload = outMsg.payload;
-    this->mQueue_Out.push(outMqttMsg);
+    mQueue_Out.push(outMqttMsg);
     return true;
+}
+void DataCenter::ExecPoll(void *argc) {
+    DataCenter *that = (DataCenter *)argc;
+    if(NULL == that){
+        exit(-1);
+    }
+
+    while (!that->recvThreadFlag) {
+        msg_mqtt_t inMsg = that->mQueue_In.pop();
+        that->objThreadPool->submitTask(std::bind(handleMsg, inMsg, std::ref(that->mQueue_Out)));
+    }
+
+}
+
+void DataCenter::ExecReponse(void *argc) {
+    DataCenter *that = (DataCenter *)argc;
+    if(NULL == that){
+        exit(-1);
+    }
+
+    while(!that->sendThreadFlag){
+        msg_mqtt_t outMsg = that->mQueue_Out.pop();
+        std::cout << "send:" << outMsg.topic << " : " << outMsg.payload << std::endl;
+        that->objmqttClient->publish(outMsg.topic,outMsg.payload);
+    }
 }

datacenter/src/thread_pool.cpp

@@ -0,0 +1,52 @@
+#include "thread_pool.h"
+
+const int TASK_MAX_NUM = 100;
+
+ThreadPool::ThreadPool():
+        initThreadNum_(4),
+        taskNum_(0),
+        taskQueMaxThreshold_(TASK_MAX_NUM),
+        stop_(false) {
+}
+
+
+ThreadPool::~ThreadPool() {
+    // 关闭线程池
+    stop_ = true;
+    notEmpty_.notify_all();
+    for (std::shared_ptr<std::thread>& t : threads_) {
+        t->join();
+    }
+}
+
+
+
+void ThreadPool::start(int initThreadNum) {
+    initThreadNum_ = initThreadNum;
+    std::cout << "[Thread Pool] Number of threads are set to " << initThreadNum_ << '.' << std::endl;
+    // 创建线程
+    for (int i=0; i < initThreadNum_; ++i){
+        std::shared_ptr<std::thread> ptr = std::make_shared<std::thread>(&ThreadPool::threadFunc, this);
+        threads_.emplace_back(std::move(ptr));
+    }
+}
+
+
+void ThreadPool::threadFunc() {
+    for (;;) {
+        std::unique_lock<std::mutex> ulock(taskQueMtx_);
+
+        notEmpty_.wait(ulock, [&]() -> bool {return !taskQue_.empty() || stop_;});
+
+        if (taskQue_.empty() && stop_) return ;
+
+        std::shared_ptr<Task> task = taskQue_.front();
+        taskQue_.pop();
+        --taskNum_;
+        notFull_.notify_one();
+
+        ulock.unlock();
+        (*task)();
+    }
+
+}