基于BOOST 实现并发服务器框架-白红宇

基于BOOST 实现并发服务器框架

阅读量：6331 次

发布时间：2019-06-22

本文共 6233 字，大约阅读时间需要 20 分钟。

一：设计思路

　　本服务器框架使用 UDP 传输协议，程序柱线程等待客户端数据，并将数组存取队列缓冲区。另外可开启多个工作线程，工作线程可以依据具体项目实现不同的功能，例如可以将队列缓冲区中的数据逐个取出存入数据库，本程序为说明方便只是将缓冲区中的数据逐个打印输出。

二：代码示例

1. 队列缓冲区实现

　　队列缓冲区由 unordered_map 和 queue 共同实现，unordered_map 散列表中每个元素 key 设置成客户端的ip value 设置成该ip 地址传过来数据的队列。

1 //file name: exclusive_queue.h  2   3 #ifndef __EXCLUSIVE_QUEUE_H__  4 #define __EXCLUSIVE_QUEUE_H__  5   6 #include 
     
        7 #include 
      
         8 #include 
       
          9 #include 
        
          10 #include 
         
           11 #include 
          
            12 #include 
           
             13 #include 
            
              14 15 #define BOOST_DATE_TIME_SOURCE 16 #define BOOST_THREAD_NO_LIB 17 18 using namespace std; 19 using namespace boost; 20 21 namespace proxy 22 { 23 24 template 
             
               25 class exclusive_queue 26 { 27 private : 28 29 typedef mutex mutex_t; 30 31 class QueueState 32 { 33 public : 34 typedef shared_ptr
              
               
                 > queue_t; 35 mutex_t m_mu_queue; 36 short m_busy; 37 queue_t m_queue; 38 39 QueueState() : m_busy(0), m_queue(new queue
                
                 ()) 40 { 41 } 42 43 }; 44 45 typedef shared_ptr
                 
                   queue_state_t; 46 typedef shared_ptr
                  
                   
                     > unordered_map_t; 47 48 unsigned int m_queue_size; 49 unordered_map_t m_queues; 50 queue_state_t m_cur_que; 51 mutex_t m_mu_queues; 52 53 public : 54 55 exclusive_queue(int cnt = 5) : m_queue_size(cnt) 56 ,m_queues(new unordered_map
                    
                     ()) 57 { 58 59 } 60 61 void Enqueue(string key, T element) 62 { 63 { 64 mutex_t::scoped_lock lock(m_mu_queues); 65 if(!m_queues->count(key)) 66 { 67 m_queues->insert(make_pair(key, make_shared
                     
                      ())); 68 } 69 m_cur_que = (*m_queues)[key]; 70 } 71 72 { 73 mutex_t::scoped_lock lock(m_cur_que->m_mu_queue); 74 if((*m_cur_que).m_queue->size() >= m_queue_size) 75 { 76 cout<<"queue is full "<< key <
                      
                       push(element); 81 } 82 } 83 84 } 85 86 void Dequeue(queue
                       
                        & result, string& key) 87 { 88 queue_state_t cur_qs; 89 BOOST_ASSERT(cur_qs == NULL); 90 while(cur_qs == NULL) 91 { 92 mutex_t::scoped_lock lock(m_mu_queues); 93 BOOST_AUTO(it, m_queues->begin()); 94 for(; it != m_queues->end(); ++it) 95 { 96 if((*it).second->m_busy == 0 && (*it).second->m_queue->size() > 0) 97 { 98 key = (*it).first; 99 cur_qs = (*it).second;100 cur_qs->m_busy = 1;101 break;102 }103 }104 105 } 106 BOOST_ASSERT(cur_qs != NULL);107 {108 if(cur_qs->m_busy == 1)109 {110 mutex_t::scoped_lock lock(cur_qs->m_mu_queue);111 while(!cur_qs->m_queue->empty())112 {113 result.push(cur_qs->m_queue->front());114 cur_qs->m_queue->pop();115 }116 BOOST_ASSERT(cur_qs->m_queue->size() == 0);117 }118 }119 120 121 }122 123 void Release(string key)124 {125 {126 mutex_t::scoped_lock lock(m_mu_queues);127 (*m_queues)[key]->m_busy = 0;128 }129 130 }131 132 133 };134 135 136 137 }138 139 #endif

2 服务器端实现

服务器主线程首先开启若干个工作线程，工作线程循环读取数据缓冲区内容，当数据缓冲器有数据则取出打印输出。

1 //file name: server.h 2  3 #ifndef __SERVER_H__ 4 #define __SERVER_H__ 5  6 #define BOOST_THREAD_NO_LIB 7 #include 
     
       8  9 #define BOOST_REGEX_NO_LIB10 #define BOOST_DATE_TIME_SOURCE11 #define BOOST_SYSTEM_NO_LIB12 #include 
      
       13 #include 
       
        14 15 #include 
        
         16 #include 
         
          17 18 #include 
          
           19 20 #include "configurationmanager.h"21 #include "datadump.h"22 #include "exclusive_queue.h"23 24 using namespace boost::asio;25 using namespace std;26 namespace proxy27 {28 29 class server30 {31 typedef shared_ptr
           
             udp_socket_t;32 private :33 io_service m_ios;34 udp_socket_t m_socket;35 ip::udp::endpoint m_remote_ep;36 system::error_code m_ec;37 proxy::exclusive_queue
            
              m_data_queue;38 39 private :40 41 void post_async(int id)42 {43 queue
             
               data_queue;44 cout<<"[Thread]:"<
              
               <<"start"<
               
                receive_from(buffer(msg.m_data), m_remote_ep, 0, m_ec);67 68 if(ec && ec != error::message_size)69 {70 throw system::system_error(ec);71 }72 msg.m_ip = ep.address().to_string();73 m_data_queue.Enqueue(ep.address().to_string(), msg);74 }75 }76 77 public :78 79 server() 80 {81 int i;82 int work_thread_num = ConfigurationManager
                
                 ::AppSetting("conf.WorkThreadNum");83 int port = ConfigurationManager
                 
                  ::AppSetting("conf.ListenningPort");84 ip::udp::endpoint local_ep(ip::udp::v4(), port);85 m_socket = udp_socket_t(new ip::udp::socket(m_ios, local_ep));86 for(i = 0; i < work_thread_num; ++i)87 {88 thread t(boost::bind(&server::post_async, this, i));89 }90 server_start();91 }92 93 94 };95 96 }97 98 99 #endif

3 数据协议格式

本程序协议格式只存在两个字段，客户端的 ip 与数据，结构参考代码

1 //file name : datamessage.h 2  3 #ifndef __DATAMESSAGE_H__ 4 #define __DATAMESSAGE_H__ 5 #include 
     
       6  7 namespace proxy 8 { 9 10 class DataMessage11 {12 13 public :14     std::string m_ip;15     char m_data[1024];16 };17 18 }19 20 #endif

4 工作线程

本程序可依据不同的项目具体实现，本程序为说明简单，只是单纯的将工作线程取到的数据全部打印输出

1 //file name: datadump.h 2  3 #ifndef __DATADUMP_H__ 4 #define __DATADUMP_H__ 5  6 #include 
     
       7 #include 
      
        8 #include 
       
         9 10 #include "datamessage.h"11 using namespace std;12 using namespace boost;13 14 namespace proxy15 {16 17 class DataDump18 {19 20 public :21     static void Dump(queue
        
          &queue)22     {23         DataMessage msg;24         while(queue.size()  != 0)25         {26             msg = queue.front();27             queue.pop();28             cout<<"[dump] "<
         
          <<" : "<<&(msg.m_data[0]) <

5 读取配置文件

　　本程序只是简单讲boost 库中的 propertytree 库进行了简单封装，想参考Boost 详细说明

1 //file name : configurationmanager.h 2  3 #ifndef __CONFIGURATIONMANAGER_H__ 4 #define __CONFIGURATIONMANAGER_H__ 5  6 #include 
     
       7 #include 
      
        8 #include 
       
         9 #include 
        
         10 #include 
         
          11 12 using namespace std;13 using namespace boost::property_tree;14 15 namespace proxy16 {17 18 template 
          
           19 class ConfigurationManager20 {21 private :22 23 public :24 static T AppSetting(string path)25 {26 ptree pt;27 read_xml("conf.xml", pt);28 29 return pt.get
           
            (path);30 }31 };32 33 34 35 }36 37 #endif

6 配置文件

　　使用 xml 格式，存放服务器监听端口与工作线程的数量

三主要技术说明

　主要基于boost 库多线程，互斥锁，网络传输套接字，重点是实现一个线程安全型的数据队列。程序结构非常简单

　　g++ *.cpp -o main -Wall -lboost_thread -lboost_system

转载于:https://www.cnblogs.com/firemage/p/4864124.html

你可能感兴趣的文章