一：设计思路
本服务器框架使用 UDP 传输协议，程序柱线程等待客户端数据，并将数组存取队列缓冲区。另外可开启多个工作线程，工作线程可以依据具体项目实现不同的功能 ，例如可以将队列缓冲区中的数据逐个取出存入数据库，本程序为说明方便只是将缓冲区中的数据逐个打印输出。
二：代码示例
1. 队列缓冲区实现
队列缓冲区由 unordered_map 和 queue 共同实现，unordered_map 散列表中每个元素 key 设置成客户端的ip value 设置成该ip 地址传过来数据的队列。
//file name: exclusive_queue.h #ifndef __EXCLUSIVE_QUEUE_H__
#define __EXCLUSIVE_QUEUE_H__ #include <boost/shared_ptr.hpp>
#include <boost/make_shared.hpp>
#include <boost/thread.hpp>
#include <boost/unordered/unordered_map.hpp>
#include <boost/typeof/typeof.hpp>
#include <boost/assert.hpp>
#include <queue>
#include <string> #define BOOST_DATE_TIME_SOURCE
#define BOOST_THREAD_NO_LIB using namespace std;
using namespace boost; namespace proxy
{ template <typename T>
class exclusive_queue
{
private : typedef mutex mutex_t; class QueueState
{
public :
typedef shared_ptr<queue<T> > queue_t;
mutex_t m_mu_queue;
short m_busy;
queue_t m_queue; QueueState() : m_busy(), m_queue(new queue<T>())
{
} }; typedef shared_ptr<QueueState > queue_state_t;
typedef shared_ptr<unordered_map<string, queue_state_t> > unordered_map_t; unsigned int m_queue_size;
unordered_map_t m_queues;
queue_state_t m_cur_que;
mutex_t m_mu_queues; public : exclusive_queue(int cnt = ) : m_queue_size(cnt)
,m_queues(new unordered_map<string, queue_state_t>())
{ } void Enqueue(string key, T element)
{
{
mutex_t::scoped_lock lock(m_mu_queues);
if(!m_queues->count(key))
{
m_queues->insert(make_pair(key, make_shared<QueueState>()));
}
m_cur_que = (*m_queues)[key];
} {
mutex_t::scoped_lock lock(m_cur_que->m_mu_queue);
if((*m_cur_que).m_queue->size() >= m_queue_size)
{
cout<<"queue is full "<< key <<endl;
}
else
{
(*m_cur_que).m_queue->push(element);
}
} } void Dequeue(queue<T>& result, string& key)
{
queue_state_t cur_qs;
BOOST_ASSERT(cur_qs == NULL);
while(cur_qs == NULL)
{
mutex_t::scoped_lock lock(m_mu_queues);
BOOST_AUTO(it, m_queues->begin());
for(; it != m_queues->end(); ++it)
{
if((*it).second->m_busy == && (*it).second->m_queue->size() > )
{
key = (*it).first;
cur_qs = (*it).second;
cur_qs->m_busy = ;
break;
}
} }
BOOST_ASSERT(cur_qs != NULL);
{
if(cur_qs->m_busy == )
{
mutex_t::scoped_lock lock(cur_qs->m_mu_queue);
while(!cur_qs->m_queue->empty())
{
result.push(cur_qs->m_queue->front());
cur_qs->m_queue->pop();
}
BOOST_ASSERT(cur_qs->m_queue->size() == );
}
} } void Release(string key)
{
{
mutex_t::scoped_lock lock(m_mu_queues);
(*m_queues)[key]->m_busy = ;
} } }; } #endif
2 服务器端实现
服务器主线程首先开启若干个工作线程， 工作线程循环读取数据缓冲区内容，当数据缓冲器有数据则取出 打印输出。
//file name: server.h #ifndef __SERVER_H__
#define __SERVER_H__ #define BOOST_THREAD_NO_LIB
#include <boost/thread.hpp> #define BOOST_REGEX_NO_LIB
#define BOOST_DATE_TIME_SOURCE
#define BOOST_SYSTEM_NO_LIB
#include <boost/asio.hpp>
#include <boost/date_time/posix_time/posix_time.hpp> #include <boost/shared_ptr.hpp>
#include <boost/make_shared.hpp> #include <ostream> #include "configurationmanager.h"
#include "datadump.h"
#include "exclusive_queue.h" using namespace boost::asio;
using namespace std;
namespace proxy
{ class server
{
typedef shared_ptr<ip::udp::socket> udp_socket_t;
private :
io_service m_ios;
udp_socket_t m_socket;
ip::udp::endpoint m_remote_ep;
system::error_code m_ec;
proxy::exclusive_queue<DataMessage> m_data_queue; private : void post_async(int id)
{
queue<DataMessage> data_queue;
cout<<"[Thread]:"<<id<<"start"<<endl;
string logstr;
string key;
while()
{
m_data_queue.Dequeue(data_queue, key);
if(data_queue.size() == )
continue;
cout<<"[thread "<<id<<"]"<<endl;
DataDump::Dump(data_queue);
m_data_queue.Release(key);
}
} void server_start()
{
cout<<"Server start..."<<endl;
DataMessage msg;
ip::udp::endpoint ep;
system::error_code ec;
while(true)
{
m_socket->receive_from(buffer(msg.m_data), m_remote_ep, , m_ec); if(ec && ec != error::message_size)
{
throw system::system_error(ec);
}
msg.m_ip = ep.address().to_string();
m_data_queue.Enqueue(ep.address().to_string(), msg);
}
} public : server()
{
int i;
int work_thread_num = ConfigurationManager<long>::AppSetting("conf.WorkThreadNum");
int port = ConfigurationManager<long>::AppSetting("conf.ListenningPort");
ip::udp::endpoint local_ep(ip::udp::v4(), port);
m_socket = udp_socket_t(new ip::udp::socket(m_ios, local_ep));
for(i = ; i < work_thread_num; ++i)
{
thread t(boost::bind(&server::post_async, this, i));
}
server_start();
} }; } #endif
3 数据协议格式
本程序协议格式只存在两个字段 ，客户端的 ip 与 数据，结构参考代码
//file name : datamessage.h #ifndef __DATAMESSAGE_H__
#define __DATAMESSAGE_H__
#include <string> namespace proxy
{ class DataMessage
{ public :
std::string m_ip;
char m_data[];
}; } #endif
4 工作线程
本程序可依据不同的项目具体实现，本程序为说明简单，只是单纯的将工作线程取到的数据全部打印输出
//file name: datadump.h #ifndef __DATADUMP_H__
#define __DATADUMP_H__ #include <ostream>
#include <queue>
#include <string> #include "datamessage.h"
using namespace std;
using namespace boost; namespace proxy
{ class DataDump
{ public :
static void Dump(queue<DataMessage> &queue)
{
DataMessage msg;
while(queue.size() != )
{
msg = queue.front();
queue.pop();
cout<<"[dump] "<<msg.m_ip<<" : "<<&(msg.m_data[]) <<endl;
}
}
}; } #endif
5 读取配置文件
本程序只是简单讲boost 库中的 propertytree 库进行了简单封装，想参考Boost 详细说明
//file name : configurationmanager.h #ifndef __CONFIGURATIONMANAGER_H__
#define __CONFIGURATIONMANAGER_H__ #include <ostream>
#include <boost/property_tree/ptree.hpp>
#include <boost/property_tree/xml_parser.hpp>
#include <boost/assign.hpp>
#include <boost/typeof//typeof.hpp> using namespace std;
using namespace boost::property_tree; namespace proxy
{ template <class T>
class ConfigurationManager
{
private : public :
static T AppSetting(string path)
{
ptree pt;
read_xml("conf.xml", pt); return pt.get<T>(path);
}
}; } #endif
6 配置文件
使用 xml 格式，存放服务器监听端口与工作线程的数量
<?xml version="1.0" encoding="utf-8"?>
<conf>
<ListenningPort></ListenningPort>
<WorkThreadNum></WorkThreadNum>
</conf>
三 主要技术说明
主要基于boost 库多线程，互斥锁，网络传输套接字，重点是实现一个线程安全型的数据队列。程序结构非常简单
g++ *.cpp -o main -Wall -lboost_thread -lboost_system