// = Active object activation method.

virtual int activate (long flags = THR_NEW_LWP,

int n_threads = 1,

int force_active = 0,

long priority = ACE_DEFAULT_THREAD_PRIORITY,

int grp_id = -1,

ACE_Task_Base *task = 0,

ACE_hthread_t thread_handles[] = 0,

void *stack[] = 0,

size_t stack_size[] = 0,

ACE_thread_t thread_names[] = 0);

　　第一個(gè)參數(shù)

THR_CANCEL_DISABLE, THR_CANCEL_ENABLE, THR_CANCEL_DEFERRED,

THR_CANCEL_ASYNCHRONOUS, THR_BOUND, THR_NEW_LWP, THR_DETACHED,

THR_SUSPENDED, THR_DAEMON, THR_JOINABLE, THR_SCHED_FIFO,

THR_SCHED_RR, THR_SCHED_DEFAULT

　　第二個(gè)參數(shù)

　　第四個(gè)參數(shù)用于設(shè)置運(yùn)行線程的優(yōu)先級(jí)。缺省情況下，或優(yōu)先級(jí)被設(shè)為

　　還可以傳入將要?jiǎng)?chuàng)建的線程的線程句柄、線程名和棧空間，以在線程創(chuàng)建過(guò)程中使用。如果它們被設(shè)置為

　　下面的例子可以幫助你進(jìn)一步理解

例

#include ”ace/Reactor.h”

#include ”ace/Svc_Handler.h”

#include ”ace/Acceptor.h”

#include ”ace/Synch.h”

#include ”ace/SOCK_Acceptor.h”

class MyServiceHandler; //forward declaration

typedef ACE_Singleton<ACE_Reactor,ACE_Null_Mutex> Reactor;

typedef ACE_Acceptor<MyServiceHandler,ACE_SOCK_ACCEPTOR> Acceptor;

class MyServiceHandler:

public ACE_Svc_Handler<ACE_SOCK_STREAM,ACE_MT_SYNCH>

{

// The two thread names are kept here

ACE_thread_t thread_names[2];

public:

int open(void*)

{

ACE_DEBUG((LM_DEBUG, ”Acceptor: received new connection \n”));

　

//Register with the reactor to remember this handler..

Reactor::instance()

->register_handler(this,ACE_Event_Handler::READ_MASK);

ACE_DEBUG((LM_DEBUG,”Acceptor: ThreadID:(%t) open\n”));

　

//Create two new threads to create and send messages to the

//remote machine.

activate(THR_NEW_LWP,

2, //2 new threads

0, //force active false, if already created don’t try again.

ACE_DEFAULT_THREAD_PRIORITY,//Use default thread priority

-1,

this,//Which ACE_Task object to create? In this case this one.

0,// don’t care about thread handles used

0,// don’t care about where stacks are created

0,//don’t care about stack sizes

thread_names); // keep identifiers in thread_names

　

//keep the service handler registered with the acceptor.

return 0;

}

void send_message1(void)

{

//Send message type 1

ACE_DEBUG((LM_DEBUG,”(%t)Sending message::>>”));

　

//Send the data to the remote peer

ACE_DEBUG((LM_DEBUG,”Sent message1”));

peer().send_n(”Message1”,LENGTH_MSG_1);

} //end send_message1

int send_message2(void)

{

//Send message type 1

ACE_DEBUG((LM_DEBUG,”(%t)Sending message::>>”));

　

//Send the data to the remote peer

ACE_DEBUG((LM_DEBUG,”Sent Message2”));

peer().send_n(”Message2”,LENGTH_MSG_2);

}//end send_message_2

　

int svc(void)

{

ACE_DEBUG( (LM_DEBUG,”(%t) Svc thread \n”));

if(ACE_Thread::self()== thread_names[0])

while(1) send_message1(); //send message 1s forever

else

while(1) send_message2(); //send message 2s forever

　

return 0; // keep the com, piler happy.

}

int handle_input(ACE_HANDLE)

{

ACE_DEBUG((LM_DEBUG,”(%t) handle_input ::”));

char* data= new char[13];

　

//Check if peer aborted the connection

if(peer().recv_n(data,12)==0)

{

printf(”P(pán)eer probably aborted connection”);

return -1; //de-register from the Reactor.

}

　

//Show what you got..

ACE_OS::printf(”<< %s\n”,data);

　

//keep yourself registered

return 0;

}

};

int main(int argc, char* argv[])

{

ACE_INET_Addr addr(10101);

ACE_DEBUG((LM_DEBUG,”Thread: (%t) main”));

//Prepare to accept connections

Acceptor myacceptor(addr,Reactor::instance());

// wait for something to happen.

while(1)

Reactor::instance()->handle_events();

　

return 0;

}

在此例中，服務(wù)處理器在它的

　　ACE_Message_Block

　　下面的例子演示怎樣使用消息隊(duì)列的一些功能：

例

#include ”ace/Reactor.h”

#include ”ace/Svc_Handler.h”

#include ”ace/Acceptor.h”

#include ”ace/Synch.h”

#include ”ace/SOCK_Acceptor.h”

#include ”ace/Thread.h”

#define NETWORK_SPEED 3

class MyServiceHandler; //forward declaration

typedef ACE_Singleton<ACE_Reactor,ACE_Null_Mutex> Reactor;

typedef ACE_Acceptor<MyServiceHandler,ACE_SOCK_ACCEPTOR> Acceptor;

class MyServiceHandler:

public ACE_Svc_Handler<ACE_SOCK_STREAM,ACE_MT_SYNCH>{

// The message sender and creator threads are handled here.

ACE_thread_t thread_names[2];

public:

int open(void*)

{

ACE_DEBUG((LM_DEBUG, ”Acceptor: received new connection \n”));

　

//Register with the reactor to remember this handler..

Reactor::instance()

->register_handler(this,ACE_Event_Handler::READ_MASK);

ACE_DEBUG((LM_DEBUG,”Acceptor: ThreadID:(%t) open\n”));

　

//Create two new threads to create and send messages to the

//remote machine.

activate(THR_NEW_LWP,

2, //2 new threads

0,

ACE_DEFAULT_THREAD_PRIORITY,

-1,

this,

0,

0,

0,

thread_names); // identifiers in thread_handles

　

//keep the service handler registered with the acceptor.

return 0;

}

void send_message(void)

{

//Dequeue the message and send it off

ACE_DEBUG((LM_DEBUG,”(%t)Sending message::>>”));

　

//dequeue the message from the message queue

ACE_Message_Block *mb;

ACE_ASSERT(this->getq(mb)!=-1);

int length=mb->length();

char *data =mb->rd_ptr();

　

//Send the data to the remote peer

ACE_DEBUG((LM_DEBUG,”%s \n”,data,length));

peer().send_n(data,length);

　

//Simulate very SLOW network.

ACE_OS::sleep(NETWORK_SPEED);

　

//release the message block

mb->release();

} //end send_message

int construct_message(void)

{

// A very fast message creation algorithm

// would lead to the need for queuing messages..

// here. These messages are created and then sent

// using the SLOW send_message() routine which is

// running in a different thread so that the message

//construction thread isn’t blocked.

ACE_DEBUG((LM_DEBUG,”(%t)Constructing message::>> ”));

　

// Create a new message to send

ACE_Message_Block *mb;

char *data=”Hello Connector”;

ACE_NEW_RETURN (mb,ACE_Message_Block (16,//Message 16 bytes long

ACE_Message_Block::MB_DATA,//Set header to data

0,//No continuations.

data//The data we want to send

), 0);

mb->wr_ptr(16); //Set the write pointer.

　

// Enqueue the message into the message queue

// we COULD have done a timed wait for enqueuing in case

// someone else holds the lock to the queue so it doesn’t block

//forever..

ACE_ASSERT(this->putq(mb)!=-1);

ACE_DEBUG((LM_DEBUG,”Enqueued msg successfully\n”));

}

int svc(void)

{

ACE_DEBUG( (LM_DEBUG,”(%t) Svc thread \n”));

　

//call the message creator thread

if(ACE_Thread::self()== thread_names[0])

while(1) construct_message(); //create messages forever

else

while(1) send_message(); //send messages forever

　

return 0; // keep the compiler happy.

}

int handle_input(ACE_HANDLE)

{

ACE_DEBUG((LM_DEBUG,”(%t) handle_input ::”));

char* data= new char[13];

　

//Check if peer aborted the connection

if(peer().recv_n(data,12)==0)

{

printf(”P(pán)eer probably aborted connection”);

return -1; //de-register from the Reactor.

}

　

//Show what you got..

ACE_OS::printf(”<< %s\n”,data);

　

//keep yourself registered

return 0;

}

};

int main(int argc, char* argv[])

{

ACE_INET_Addr addr(10101);

ACE_DEBUG((LM_DEBUG,”Thread: (%t) main”));

//Prepare to accept connections

Acceptor myacceptor(addr,Reactor::instance());

// wait for something to happen.

while(1)

Reactor::instance()->handle_events();

　

return 0;

}

這個(gè)例子演示怎樣使用

　　要了解更多關(guān)于如何使用消息塊、數(shù)據(jù)塊或是消息隊(duì)列的信息，請(qǐng)閱讀此教程中有關(guān)“消息隊(duì)列”、

　　接受器和連接器工廠（也就是

　　在使用接受器的情況下，應(yīng)用級(jí)程序員可以調(diào)用

　　在使用連接器的情況中，應(yīng)用程序員調(diào)用連接器工廠的

//Synchronous

OurConnector.connect_n(NUMBER_CONN,ArrayofMySvcHandlers,Remote_Addr,0,

ACE_Synch_Options::synch);

　

//Asynchronous

OurConnector.connect_n(NUMBER_CONN,ArrayofMySvcHandlers,Remote_Addr,0,

ACE_Synch_Options::asynch);

如果連接請(qǐng)求是異步的，

　　在有連接請(qǐng)求在指定的地址和端口上到來(lái)時(shí)，反應(yīng)堆自動(dòng)回調(diào)

　　該方法是一個(gè)“模板方法”（

　　在我們的這個(gè)案例中，模板方法將算法定義如下：

• make_svc_handler()

　　這些方法都可以被重新編寫(xiě)，從而靈活地決定這些操作怎樣來(lái)實(shí)際執(zhí)行。

　　這樣，

　　在服務(wù)處理器被創(chuàng)建后，

　　應(yīng)用發(fā)出的

　　在我們的這個(gè)案例中，模板方法

• make_svc_handler()

　　每一方法都可以被重新編寫(xiě)，從而靈活地決定這些操作怎樣來(lái)實(shí)際執(zhí)行。

　　這樣，在應(yīng)用發(fā)出

　　在服務(wù)處理器被創(chuàng)建后，

　　如果連接被指定為異步的，在向底層的具體連接器發(fā)出非阻塞式

　　與

　　一旦服務(wù)處理器被創(chuàng)建、連接被建立，以及句柄在服務(wù)處理器中被設(shè)置，

　　如上面所提到的，因?yàn)槭褂昧丝梢灾剌d的橋接方法，很容易對(duì)接受器和連接器進(jìn)行調(diào)諧。橋接方法允許調(diào)諧：

　　如上所示，調(diào)諧是通過(guò)重載

　　為了方便上面所提到的對(duì)接受器和連接器模式的調(diào)諧方法，

　　策略模式被用于使算法行為與類的接口去耦合。其基本概念是允許一個(gè)類（稱為

　　在

　　在

　　　　　　　　　表

　　下面的例子演示策略接受器和連接器類的使用。

例

#include ”ace/Reactor.h”

#include ”ace/Svc_Handler.h”

#include ”ace/Acceptor.h”

#include ”ace/Synch.h”

#include ”ace/SOCK_Acceptor.h”

#define PORT_NUM 10101

#define DATA_SIZE 12

//forward declaration

class My_Svc_Handler;

// Define the Service Handler

class My_Svc_Handler:

public ACE_Svc_Handler <ACE_SOCK_STREAM,ACE_NULL_SYNCH>

{

private:

char* data;

public:

My_Svc_Handler()

{

data= new char[DATA_SIZE];

}

My_Svc_Handler(ACE_Thread_Manager* tm)

{

data= new char[DATA_SIZE];

}

int open(void*)

{

cout<<”Connection established”<<endl;

　

//Register with the reactor

ACE_Event_Handler::READ_MASK);

　

return 0;

}

int handle_input(ACE_HANDLE)

{

peer().recv_n(data,DATA_SIZE);

ACE_OS::printf(”<< %s\n”,data);

　

// keep yourself registered with the reactor

return 0;

}

};

int main(int argc, char* argv[])

{

ACE_INET_Addr addr(PORT_NUM);

//Instantiate the acceptor

MyAcceptor acceptor(addr, //address to accept on

ACE_Reactor::instance(), //the reactor to use

0,

0, // don’t care about connection estb. strategy

while(1) /* Start the reactor’s event loop */

ACE_Reactor::instance()->handle_events();

}

這個(gè)例子基于上面的例

　　在許多應(yīng)用中，客戶會(huì)連接到服務(wù)器，然后重新連接到同一服務(wù)器若干次；每次都要建立連接，執(zhí)行某些工作，然后掛斷連接（比如像在

　　當(dāng)客戶試圖重新建立連接到先前已經(jīng)連接的服務(wù)器時(shí)，

例

#include ”ace/Reactor.h”

#include ”ace/Svc_Handler.h”

#include ”ace/Connector.h”

#include ”ace/Synch.h”

#include ”ace/SOCK_Connector.h”

#include ”ace/INET_Addr.h”

#define PORT_NUM 10101

#define DATA_SIZE 16

//forward declaration

class My_Svc_Handler;

//Function prototype

static void make_connections(void *arg);

// Template specializations for the hashing function for the

// hash_map which is used by the cache. The cache is used internally by the

// Cached Connection Strategy . Here we use ACE_Hash_Addr

// as our external identifier. This utility class has already

// overloaded the == operator and the hash() method. (The

// hashing function). The hash() method delegates the work to

// hash_i() and we use the IP address and port to get a

// a unique integer hash value.

size_t ACE_Hash_Addr<ACE_INET_Addr>::hash_i (const ACE_INET_Addr &addr) const

{

return addr.get_ip_address () + addr.get_port_number ();

}

//instantiate a strategy acceptor

typedef ACE_Strategy_Connector<My_Svc_Handler,ACE_SOCK_CONNECTOR>

STRATEGY_CONNECTOR;

//Instantiate the Creation Strategy

typedef ACE_NOOP_Creation_Strategy<My_Svc_Handler>

NULL_CREATION_STRATEGY;

//Instantiate the Concurrency Strategy

typedef ACE_NOOP_Concurrency_Strategy<My_Svc_Handler>

NULL_CONCURRENCY_STRATEGY;

//Instantiate the Connection Strategy

typedef ACE_Cached_Connect_Strategy<My_Svc_Handler,

ACE_SOCK_CONNECTOR,

ACE_SYNCH_RW_MUTEX>

CACHED_CONNECT_STRATEGY;

class My_Svc_Handler:

public ACE_Svc_Handler <ACE_SOCK_STREAM,ACE_MT_SYNCH>

{

private:

char* data;

public:

My_Svc_Handler()

{

data= new char[DATA_SIZE];

}

My_Svc_Handler(ACE_Thread_Manager* tm)

{

data= new char[DATA_SIZE];

}

//Called before the service handler is recycled..

int

{

ACE_DEBUG ((LM_DEBUG,

”(%P|%t) recycling Svc_Handler %d with handle %d\n”,

this, this->peer ().get_handle ()));

return 0;

}

int open(void*)

{

ACE_DEBUG((LM_DEBUG,”(%t)Connection established \n”));

　

//Register the service handler with the reactor

ACE_Reactor::instance()

->register_handler(this,ACE_Event_Handler::READ_MASK);

　

activate(THR_NEW_LWP|THR_DETACHED);

return 0;

}

int handle_input(ACE_HANDLE)

{

ACE_DEBUG((LM_DEBUG,”Got input in thread: (%t) \n”));

peer().recv_n(data,DATA_SIZE);

ACE_DEBUG((LM_DEBUG,”<< %s\n”,data));

　

//keep yourself registered with the reactor

return 0;

}

int svc(void)

{

//send a few messages and then mark connection as idle so that it can

// be recycled

ACE_DEBUG((LM_DEBUG,”Started the service routine \n”));

for(int i=0;i<3;i++)

{

ACE_DEBUG((LM_DEBUG,”(%t)>>Hello World\n”));

ACE_OS::fflush(stdout);

peer().send_n(”Hello World”,sizeof(”Hello World”));

}

//Mark the service handler as being idle now and let the

//other threads reuse this connection

this->idle(1);

//Wait for the thread to die

this->thr_mgr()->wait();

return 0;

}

};

ACE_INET_Addr *addr;

int main(int argc, char* argv[])

{

addr= new ACE_INET_Addr(PORT_NUM,argv[1]);

//Creation Strategy

NULL_CREATION_STRATEGY creation_strategy;

//Concurrency Strategy

NULL_CONCURRENCY_STRATEGY concurrency_strategy;

//Connection Strategy

CACHED_CONNECT_STRATEGY caching_connect_strategy;

//instantiate the connector

STRATEGY_CONNECTOR connector(

ACE_Reactor::instance(), //the reactor to use

&creation_strategy,

&caching_connect_strategy,

&concurrency_strategy);

　

//Use the thread manager to spawn a single thread

//to connect multiple times passing it the address

//of the strategy connector

if(ACE_Thread_Manager::instance()->spawn(

(ACE_THR_FUNC) make_connections,

(void *) &connector,

THR_NEW_LWP) == -1)

　

ACE_ERROR ((LM_ERROR, ”(%P|%t) %p\n%a”, ”client thread spawn failed”));

while(1) /* Start the reactor’s event loop */

ACE_Reactor::instance()->handle_events();

}

//Connection establishment function, tries to establish connections

//to the same server again and re-uses the connections from the cache

void make_connections(void *arg)

{

ACE_DEBUG((LM_DEBUG,”(%t)Prepared to connect \n”));

STRATEGY_CONNECTOR *connector= (STRATEGY_CONNECTOR*) arg;

for (int i = 0; i < 10; i++)

{

My_Svc_Handler *svc_handler = 0;

　

// Perform a blocking connect to the server using the Strategy

// Connector with a connection caching strategy. Since we are

// connecting to the same <server_addr> these calls will return the

// same dynamically allocated <Svc_Handler> for each <connect> call.

if (connector->connect (svc_handler, *addr) == -1)

{

ACE_ERROR ((LM_ERROR, ”(%P|%t) %p\n”, ”connection failed\n”));

return;

}

　

// Rest for a few seconds so that the connection has been freed up

ACE_OS::sleep (5);

}

在上面的例子中，緩存式連接策略被用于緩存連接。要使用這一策略，需要一點(diǎn)額外的工作：定義

　　這個(gè)例子比至今為止我們所列舉的例子都要復(fù)雜一點(diǎn)，所以有理由多進(jìn)行一點(diǎn)討論。

　　和平常一樣，一旦連接建立后，反應(yīng)堆回調(diào)我們的服務(wù)處理器（

　　在

　　有時(shí), ，使用重量級(jí)的

例

#include ”ace/Reactor.h”

#include ”ace/Svc_Handler.h”

#include ”ace/Acceptor.h”

#include ”ace/Synch.h”

#include ”ace/SOCK_Acceptor.h”

#define PORT_NUM 10101

#define DATA_SIZE 12

//forward declaration

class My_Event_Handler;

//Create the Acceptor class

typedef ACE_Acceptor<My_Event_Handler,ACE_SOCK_ACCEPTOR>

MyAcceptor;

//Create an event handler similar to as seen in example 2. We have to

//overload the get_handle() method and write the peer() method. We also

//provide the data member peer_ as the underlying stream which is

//used.

class My_Event_Handler: public ACE_Event_Handler

{

private:

char* data;

//Add a new attribute for the underlying stream which will be used by

//the Event Handler

ACE_SOCK_Stream peer_;

public:

My_Event_Handler()

{

data= new char[DATA_SIZE];

}

int open(void*)

{

cout<<”Connection established”<<endl;

　

//Register the event handler with the reactor

ACE_Reactor::instance()->register_handler(this,

ACE_Event_Handler::READ_MASK);

　

return 0;

}

int handle_input(ACE_HANDLE)

{

// After using the peer() method of our ACE_Event_Handler to obtain a

//reference to the underlying stream of the service handler class we

//call recv_n() on it to read the data which has been received. This

//data is stored in the data array and then printed out

peer().recv_n(data,DATA_SIZE);

ACE_OS::printf(”<< %s\n”,data);

　

// keep yourself registered with the reactor

return 0;

}

// new method which returns the handle to the reactor when it

//asks for it.

ACE_HANDLE get_handle(void) const

{

return this->peer_.get_handle();

}

//new method which returns a reference to the peer stream

ACE_SOCK_Stream &peer(void) const

{

return (ACE_SOCK_Stream &) this->peer_;

}

};

int main(int argc, char* argv[])

{

ACE_INET_Addr addr(PORT_NUM);

//create the acceptor

MyAcceptor acceptor(addr, //address to accept on

ACE_Reactor::instance()); //the reactor to use

while(1) /* Start the reactor’s event loop */

ACE_Reactor::instance()->handle_events();

}