C++之ACE实现通用服务器的C/S架构通信程序

C++ACE实现通用的服务器C/S架构通信程序
ACE自适配通信环境（ADAPTIVE Communication Environment）是可以自由使用、
开放源码的面向对象（OO）框架（Framework），在其中实现了许多用于并发通信软件的核心模式。
ACE提供了一组丰富的可复用C++ Wrapper Facade（包装外观）和框架组件，
可跨越多种平台完成通用的通信软件任务，其中包括：
事件多路分离和事件处理器分派、信号处理、服务初始化、进程间通信、共享内存管理、消息路由、
分布式服务动态（重）配置、并发执行和同步，等等。
ACE的目标用户是高性能和实时通信服务和应用的开发者。
它简化了使用进程间通信、事件多路分离、显式动态链接和并发的OO网络应用和服务的开发。
此外，通过服务在运行时与应用的动态链接，ACE还使系统的配置和重配置得以自动化。

我最近采用ACE实现了一个通用的C/S架构通信程序，具体实现简述如下：
1． 服务器端：一个采用领导者/跟随者模型的线程池不断地接受从多个客户端发来的消息，
并放入一个消息队列，然后又有一个采用半同步/半异步模型的线程池不断地从这个消息队列中取出消息进行处理。
服务器端代码如下：（共10个文件）

//ACE_Client.cpp
#include "Client.h"
using namespace ACE_Client;
int main(int argc, char *argv[])
{
Client client("localhost"); //服务器的IP地址或者服务器名称
for(int i = 0; i < 5; i++)
{
char *task1 = "Is it a good day?"; //第1个task的数据
size_t task1_len = 18; //第1个task的数据长度
char *task1_t; //无需修改
ACE_NEW_RETURN(task1_t, char[task1_len + 4], -1); //无需修改
client.put_task(task1_t, task1, task1_len); //无需修改
char *task2 = "Yeah, it really is."; //第2个task的数据
size_t task2_len = 20; //第2个task的数据长度
char *task2_t; //无需修改
ACE_NEW_RETURN(task2_t, char[task2_len + 4], -1); //无需修改
client.put_task(task2_t, task2, task2_len); //无需修改
client.send_tasks(); //将上面的task全部发到服务器
delete [] task1_t; //释放task1的内存
delete [] task2_t; //释放task2的内存
}
return 0;
}
------------------------------------------------------------------------
Constants.h
#ifndef __CONSTANTS_H_
#define __CONSTANTS_H_
#include "ace/Time_Value.h"
namespace ACE_Client
{
static const size_t BUFFER_SIZE = 4096; //数据缓冲区大小
static const size_t SERVER_PORT_NUM = 10101; //服务器的通信端口号
static const ACE_Time_Value TIME_INTERVAL(0, 1000000); //两次数据发送之间的时间间隔(0 s + 1000000 us = 1 s)
}
#endif
------------------------------------------------------------------------
Client.h
#ifndef __CLIENT_H_
#define __CLIENT_H_
#include "ace/Task.h"
#include "ace/INET_Addr.h"
#include "ace/Synch.h"
namespace ACE_Client
{
class Client: public ACE_Task<ACE_NULL_SYNCH>
{
public:
Client(char *server);
virtual int svc(void);
char *put_task(char *msg_t, char *msg_s, size_t msg_len);
void send_tasks(void);
private:
ACE_INET_Addr addr;
};
}
#endif
------------------------------------------------------------------------
Client.cpp
#include "ace/OS.h"
#include "ace/SOCK_Stream.h"
#include "ace/SOCK_Connector.h"
#include "ace/Message_Block.h"
#include "ace/Thread_Manager.h"
#include "ace/INET_Addr.h"
#include "Constants.h"
#include "Client.h"
namespace ACE_Client
{
Client::Client(char *server)
{
addr = ACE_INET_Addr(SERVER_PORT_NUM, server);
}
int Client::svc(void)
{
ACE_SOCK_Stream stream;
ACE_SOCK_Connector connector;
if(connector.connect(stream, addr) < 0)
{
return -1;
}
else
{
while(1)
{
ACE_Message_Block *mb = NULL;
if(this->getq(mb) == -1)
{
break;
}
ACE_DEBUG((LM_DEBUG, ACE_TEXT("(%t) Sending %s\n"), &(mb->rd_ptr()[4])));
stream.send_n(mb->rd_ptr(), mb->length());
mb->release();
ACE_OS::sleep(TIME_INTERVAL);
}
}
stream.close();
return 0;
}
char *Client::put_task(char *msg_t, char *msg_s, size_t msg_len)
{
for(int i = 0; i < 4; i++)
{
msg_t[i] = (char)((msg_len >> (8 * i)) & 0xff);
}
ACE_OS::memcpy(&msg_t[4], msg_s, msg_len);
ACE_Message_Block *mb = NULL;
ACE_NEW_RETURN(mb, ACE_Message_Block(msg_len + 4, ACE_Message_Block::MB_DATA, 0, msg_t), 0);
mb->wr_ptr(msg_len + 4);
this->putq(mb);
return msg_t;
}
void Client::send_tasks(void)
{
this->activate();
ACE_Thread_Manager::instance()->wait();
}
}　

View Code