[转帖]采用C++的ACE库实现的一个通用的udp通信服务器程序

peakzhang · 发表于 2008-7-13 22:43:55

转自：[url=http://blog.csdn.net/itclock/archive/2006/08/08/1036647.aspx]http://blog.csdn.net/itclock/archive/2006/08/08/1036647.aspx[/url]
作者：itclock

采用C++的ACE库实现的一个通用的udp通信服务器程序
全部源代码如下：

ACE_Server.cpp
#include "ace/SOCK_Acceptor.h"
#include "ace/Acceptor.h"
#include "ace/Thread_Manager.h"
#include "ace/TP_Reactor.h"
#include "ace/Reactor.h"
#include "ace/INET_Addr.h"
#include "ace/OS.h"
#include "Request_Handler.h"
#include "Server.h"
#include "Constants.h"
using namespace ACE_Server;
int main(int argc, char *argv[])
{
ACE_INET_Addr local_addr(SERVER_PORT_NUM);
Request_Handler *endpoint;
ACE_NEW_RETURN (endpoint,
Request_Handler (local_addr),
-1);
// Read data from other side.
if (ACE_Reactor::instance ()->register_handler
(endpoint,
ACE_Event_Handler::READ_MASK) == -1)
ACE_ERROR_RETURN ((LM_ERROR,
"ACE_Reactor::register_handler"),
-1);
Server server_tp;
server_tp.activate(THR_NEW_LWP | THR_JOINABLE, SERVER_THREAD_POOL_SIZE);
ACE_Thread_Manager::instance()->wait();
return 0;
}
Constants.h
#ifndef __CONSTANTS_H_
#define __CONSTANTS_H_
namespace ACE_Server
{
static const size_t SERVER_THREAD_POOL_SIZE = 5; //进行数据接收的线程池大小
static const size_t TASK_THREAD_POOL_SIZE = 5; //进行数据处理的线程池大小
static const size_t BUFFER_SIZE = 4096; //数据缓冲区大小
static const size_t SERVER_PORT_NUM = 10101; //服务器的通信端口号
}
#endif
Server.h
#ifndef __SERVER_H_
#define __SERVER_H_
#include "ace/Task.h"
namespace ACE_Server
{
class Server: public ACE_Task_Base
{
public:
virtual int svc(void);
};
}
#endif
Server.cpp
#include "ace/Reactor.h"
#include "Server.h"
namespace ACE_Server
{
int Server::svc(void)
{
int result = ACE_Reactor::instance()->run_reactor_event_loop();
if(result == -1)
return -1;
return 0;
}
}
Request_Handler.h
#ifndef __REQUEST_HANDLER_H_
#define __REQUEST_HANDLER_H_
#include "ace/Svc_Handler.h"
#include "ace/SOCK_Stream.h"
#include "ace/SOCK_Dgram.h"
#include "ace/Synch.h"
#include "ace/Thread_Manager.h"
#include "Task_Manager.h"
namespace ACE_Server
{
class Request_Handler: public ACE_Event_Handler
{
public:
Request_Handler(const ACE_INET_Addr &local_addr);
virtual ACE_HANDLE get_handle (void) const;
virtual int handle_close (ACE_HANDLE handle,
ACE_Reactor_Mask close_mask);
virtual int handle_timeout (const ACE_Time_Value & tv,
const void *arg = 0);
protected:
virtual int handle_input(ACE_HANDLE fd = ACE_INVALID_HANDLE);
private:
static Task_Manager task_mgr;
ACE_SOCK_Dgram endpoint_;
};
}
#endif
Request_Handler.cpp
#include "ace/OS.h"
#include "ace/Message_Block.h"
#include "ace/Thread_Manager.h"
#include "ace/Svc_Handler.h"
#include "ace/SOCK_Stream.h"
#include "ace/Synch.h"
#include "ace/Reactor.h"
#include "Request_Handler.h"
#include "Task_Manager.h"
#include "Constants.h"
namespace ACE_Server
{
Task_Manager Request_Handler::task_mgr;
Request_Handler::Request_Handler(const ACE_INET_Addr &local_addr)
: endpoint_ (local_addr)
{
//this->reactor(ACE_Reactor::instance());
task_mgr.activate();
}
ACE_HANDLE Request_Handler::get_handle (void) const
{
return this->endpoint_.get_handle ();
}
int Request_Handler::handle_close (ACE_HANDLE handle,
ACE_Reactor_Mask)
{
ACE_UNUSED_ARG (handle);
ACE_DEBUG ((LM_DEBUG,
"(%P|%t) handle_close\n"));
this->endpoint_.close ();
delete this;
return 0;
}
int Request_Handler::handle_timeout (const ACE_Time_Value &,
const void *)
{
ACE_DEBUG ((LM_DEBUG,
"(%P|%t) timed out for endpoint\n"));
return 0;
}
int Request_Handler::handle_input(ACE_HANDLE fd)
{
char buf[BUFSIZ];
ACE_INET_Addr from_addr;
ACE_DEBUG ((LM_DEBUG,
"(%P|%t) activity occurred on handle %d!\n",
this->endpoint_.get_handle ()));
ssize_t n = this->endpoint_.recv (buf,
sizeof buf,
from_addr);
if (n == -1)
ACE_ERROR ((LM_ERROR,
"%p\n",
"handle_input"));
else
{
ACE_DEBUG ((LM_DEBUG,
"(%P|%t) buf of size %d = %*s\n",
n,
n,
buf));
ACE_Message_Block *mb = NULL;
ACE_NEW_RETURN(mb, ACE_Message_Block(n, ACE_Message_Block::MB_DATA, 0, buf), -1);
mb->wr_ptr(n);
task_mgr.putq(mb);
}
return 0;
}
}
Task_Manager.h
#ifndef __TASK_MANAGER_H_
#define __TASK_MANAGER_H_
#include "ace/Task.h"
#include "ace/Synch.h"
namespace ACE_Server
{
class Task_Manager: public ACE_Task<ACE_MT_SYNCH>
{
public:
virtual int svc(void);
};
}
#endif
Task_Manager.cpp
#include "ace/Message_Block.h"
#include "Task_Manager.h"
#include "Task_Worker.h"
#include "Constants.h"
namespace ACE_Server
{
int Task_Manager::svc(void)
{
Task_Worker task_tp;
task_tp.activate(THR_NEW_LWP | THR_JOINABLE, TASK_THREAD_POOL_SIZE);
while(1)
{
ACE_Message_Block *mb = NULL;
if(this->getq(mb) < 0)
{
task_tp.msg_queue()->deactivate();
task_tp.wait();
}
task_tp.putq(mb);
}
return 0;
}
}
Task_Worker.h
#ifndef __TASK_WORKER_H_
#define __TASK_WORKER_H_
#include "ace/Task.h"
#include "ace/Synch.h"
#include "ace/Message_Block.h"
namespace ACE_Server
{
class Task_Worker: public ACE_Task<ACE_MT_SYNCH>
{
public:
virtual int svc(void);
private:
void process_task(ACE_Message_Block *mb);
};
}
#endif
Task_Worker.cpp
#include "ace/OS.h"
#include "ace/Message_Block.h"
#include "Task_Worker.h"
namespace ACE_Server
{
int Task_Worker::svc(void)
{
while(1)
{
ACE_Message_Block *mb = NULL;
if(this->getq(mb) == -1)
{
continue;
}
process_task(mb);
}
return 0;
}
void Task_Worker::process_task(ACE_Message_Block *mb)
{
//进行数据处理,数据的起始地址为mb->rd_ptr(),长度为mb->length()
ACE_DEBUG((LM_DEBUG, ACE_TEXT("(%t) Processing task: %s length %d\n"), mb->rd_ptr(), mb->length()));
ACE_OS::sleep(3); //模拟数据处理过程
mb->release();
}
}

复制代码

注意：代码请先读懂再测试，不要贸然使用，因为可能隐藏bug.

		自动登录	找回密码
密码			用户注册