1、Activemq-cpp開發手冊丁靖2008-05-061 引言1.1 編寫目的快速學習CMS，提高CMS開發效率，提供一個CMS開發參考手冊詳細API手冊請參考1.2 功能介紹Activemq-cpp是一個與ActiveMQ交互通訊的C+ API開發庫，為C+開發者提供了一個訪問ActiveMQ的接口。Winkeemq-cpp是一個在Activemq-cpp基礎上封裝的API庫，對一些重復機械的初始化及銷毀清除及一些不關心的細節進行了封裝，從而簡化了編程。1.3 術語解析ActiveMQ ：開源的消息隊列服務器Broker ：消息中介，每個消息隊列服務器中至少有一個broker，是消息隊列的

2、載體Destination ：消息在broker上的目的地Queue ：消息隊列 Topic ：主題 Message ：消息 Producer ：消息產生者 Consumer ：消息消費者Client ：客戶端，生產者和消費者都在客戶端上 Server ：Activemq服務器BrokerUri ：客戶端訪問服務器上broker時的Uri其它資料請參考2 開發前準備在開發前必須先安裝activemq-cpp及winkeemq-cpp庫，具體步驟參考activemq-cpp安裝及使用文檔.doc3 CMS 3.1 概述CMS（stands for C+ Messaging Service）是一組

3、C+應用程序接口(C+ API)，它提供創建、發送、接收、讀取消息的服務。定義了一組和Sun公司和它的合作伙伴設計的CMS API相同的公共應用程序接口和相應語法，使得C+程序能夠和其他消息組件進行通信。       CMS是一種與廠商無關的 API，用來訪問消息收發系統。它類似于 JDBC (Java Database Connectivity)：這里，JDBC 是可以用來訪問許多不同關系數據庫的 API，而 CMS 則提供同樣與廠商無關的訪問方法，以訪問消息收發服務。CMS 使您能夠通過消息收發服務（有時稱為消息中介程序或路由器）從一個 C

4、MS 客戶機向另一個客戶機發送消息。消息是 CMS 中的一種類型對象，由兩部分組成：報頭和消息主體。報頭由路由信息以及有關該消息的元數據組成。消息主體則攜帶著應用程序的數據或有效負載。根據有效負載的類型來劃分，可以將消息分為幾種類型，它們分別攜帶：簡單文本 (TextMessage)、可序列化的對象 (ObjectMessage)、屬性集合 (MapMessage)、字節流 (BytesMessage)、原始值流 (StreamMessage)，還有無有效負載的消息 (Message)。       消息收發系統是異步的，也就是說，CMS 客戶

5、機可以發送消息而不必等待回應。比較可知，這完全不同于基于 RPC 的（基于遠程過程的）系統，如 EJB 1.1、CORBA 和 Java RMI 的引用實現。在 RPC 中，客戶機調用服務器上某個分布式對象的一個方法。在方法調用返回之前，該客戶機被阻塞；該客戶機在可以執行下一條指令之前，必須等待方法調用結束。在 CMS 中，客戶機將消息發送給一個虛擬通道（主題或隊列），而其它 CMS 客戶機則預訂或監聽這個虛擬通道。當 CMS 客戶機發送消息時，它并不等待回應。它執行發送操作，然后繼續執行下一條指令。消息可能最終轉發到一個或許多個客戶機，這些客戶機都不需要作出回應。 CMS的通用接口集合以異步

6、方式發送或接收消息。異步方式接收消息顯然是使用間斷網絡連接的客戶 機，諸如移動電話和PDA的最好的選擇。另外， CMS采用一種寬松結合方式整合企業系統的方法，其主要的目的就是創建能夠使用跨平臺數據信息的、可移植的企業級應用程序，而把開發人力解放出來。CMS消息服務支持兩種消息模型：Point-to-Point消息(P2P)和發布訂閱消息（Publish Subscribe messaging，簡稱Pub/Sub）。CMS規范并不要求供應商同時支持這兩種消息模型，但開發者應該熟悉這兩種消息模型的優勢與缺點。P2P消息模型是在點對點之間傳遞消息時使用。如果應用程序開發者希望每一條消息都能夠被處理，

7、那么應該使用P2P消息模型。與Pub/Sub消息模型不同，P2P消息總是能夠被傳送到指定的位置。Pub/Sub模型在一到多的消息廣播時使用。如果一定程度的消息傳遞的不可靠性可以被接受的話，那么應用程序開發者也可以使用Pub/Sub消息模型。換句話說，它適用于所有的消息消費程序并不要求能夠收到所有的信息或者消息消費程序并不想接收到任何消息的情況。CMS通過允許創建持久訂閱來簡化時間相關性，即使消息預訂者未激活也可以接收到消息。此外，使用 持久訂閱還可通過隊列提供靈活性和可靠性，而仍然允許消息被發給許多的接收者。 Topic Subscriber topic Subscriber = topicS

8、ession.createDurableSubscriber(topic, subscriptionName); Connection對象表示了到兩種消息模型中的任一種的消息系統的連接。服務器端和客戶機端對象要求管理創建的CMS連接的狀態。連接是由 Connection Factory創建的并且通過JNDI查尋定位。 /取得用于 P2P的 QueueConnectionFactory QueueConnectionFactory = queueConnectionFactory(); Context messaging = new InitialContext(); QueueConnecti

9、onFactory = (QueueConnectionFactory) Messaging.lookup(“QueueConnectionFactory”); /取得用于 pub/sub的 TopicConnectionFactory TopicConnectonFactory topicConnectionFactory; Context messaging = new InitialContext(); topicConnectionFactory= (TopicConnectionFactory)messaging.lookup(“TopicConnectionFactory”); 注

10、意：用于P2P的代碼和用于PublishSubscribe的代碼非常相似。如果session被標記為transactional的話，確認消息就通過確認和校正來自動地處理。如果session沒有標記為 transactional，你有三個用于消息確認的選項。· AUTO_ACKNOWLEDGE session將自動地確認收到一則消息。· CLIENT_ACKNOWLEDGE 客戶端程 序將確認收到一則消息，調用這則消息的確認方法。 · DUPS_OK_ACKNOWLEDGE 這個選項命令session“懶散的”確認消息傳遞，可以想到，這將導致消息提供者傳遞的一些復制

11、消息可能會出錯。這種確認的方式只應當用于消息消費程序 可以容忍潛在的副本消息存在的情況。 queueSession =queueConnection.createQueueSession(false,session.AUTO_ACKNOWLEDGE);/P2P topicSession = topicConnection.createTopicSession(false, session.AUTO_ACKNOWLEDGE); /Pub-Sub注意：在本例中，一個session目的從連結中創建，非值指出session是non-transactional的，并且 session將自動地確認收到一則

12、消息。CMS現在有兩種傳遞消息的方式。標記為NON_PERSISTENT的消息最多投遞一次，而標記 為PERSISTENT的消息將使用暫存后再轉送的機理投遞。如果一個CMS服務離線，那么持久性消息不會丟失但是得等到這個服務恢復聯機時才會被傳遞。 所以默認的消息傳遞方式是非持久性的。即使使用非持久性消息可能降低內務和需要的存儲器，并且這種傳遞方式只有當你不需要接收所有的消息時才使用。雖然 CMS規范并不需要CMS供應商實現消息的優先級路線，但是它需要遞送加快的消息優先于普通級別的消息。CMS定義了從0到9的優先級路線級別，0是最低 的優先級而9則是最高的。更特殊的是0到4是正常優先級的變化幅度，

13、而5到9是加快的優先級的變化幅度。舉例來說： topicPublisher.publish (message, DeliveryMode.PERSISTENT, 8, 10000); /Pub-Sub 或 queueSender.send(message, DeliveryMode.PERSISTENT, 8, 10000);/P2P 這個代碼片斷，有兩種消息模型，映射遞送方式是持久的，優先級為加快型，生存周期是10000 (以毫秒度量 )。如果生存周期設置為零，這則消息將永遠不會過期。當消息需要時間限制否則將使其無效時，設置生存周期是有用的。CMS定義了五種不同的消息正文格式，以及調用的消息

14、類型，允許你發送并接收以一些不同形式的數據，提供現有消息格式的一些級別的兼容性。· StreamMessage - Java原始值的數據流 · MapMessage-一套名稱-值對· TextMessage-一個字符串對象· ObjectMessage-一個序列化的 Java對象· BytesMessage-一個未解釋字節的數據流CMS應用程序接口提供用于創建每種類型消息和設置荷載的方法例如，為了在一個隊列創建并發送一個 TextMessage實例，你可以使用下列語句： TextMessage message = queueSession.cr

15、eateTextMessage(); message.setText(textMsg); 以異步方式接收消息，需要創建一個消息監聽器然后注冊一個或多個使用MessageConsumer的CMS MessageListener接口。會話(主題或隊列)負責產生某些消息，這些消息被傳送到使用onMessage方法的監聽者那里。 Using namespace cms; class ExampleListener : public MessageListener /把消息強制轉化為TextMessage格式 public void onMessage(Message message) TextMess

16、age textMsg = null; / 打開并處理這段消息 當我們創建QueueReceiver和TopicSubscriber時，我們傳遞消息選擇器字符串： /P2P QueueReceiver QueueReceiver receiver; receiver = session.createReceiver(queue, selector); /Pub-Sub TopicSubscriber TopicSubscriber subscriber; subscriber = session.createSubscriber(topic, selector);為了啟動消息的交付，不論是Pu

17、b/Sub還是P2P，都需要調用start方法。 TopicConnection.start(); /pub-sub QueueConnection.start(); /P2P 當一條消息被捕捉時，這條消息做為一條必須被強制轉化為適當消息類型的普通Message對象到達。如TextMessagevoid onMessage(const Message* message) TextMessage txtMsg=dynamic_cast<TextMessage* >(message); /對 txtMsg做一些處理停止消息的傳遞，無論是Pub/Sub還是P2P，都調用stop方法。 T

18、opicConnection. stop (); /pub-sub QueueConnection. stop (); /P2P 3.2 接口描述CMS 支持兩種消息類型P2P 和Pub/Sub，分別稱作：P2P Domain 和Pub/Sub Domain，這兩種接口都繼承統一的CMS Parent 接口，CMS 主要接口如下所示：CMS Parent  ConnectionFactoryConnectionDestinationSessionMessageProducerMessageConsumer      以下是對

19、這些接口的簡單描述：    ConnectionFactory ：連接工廠，CMS 用它創建連接    Connection ：CMS 客戶端到CMS Provider 的連接    Destination ：消息的目的地    Session： 一個發送或接收消息的線程    MessageProducer： 由Session 對象創建的用來發送消息的對象    MessageConsumer： 由Session

20、對象創建的用來接收消息的對象3.3 CMS消息模型CMS 消息由以下幾部分組成：消息頭，屬性，消息體。    消息頭(Header) - 消息頭包含消息的識別信息和路由信息，消息頭包含一些標準的屬性如：CMSDestination,CMSMessageID 等。 消息頭由誰設置CMSDestinationsend 或 publish 方法CMSDeliveryModesend 或 publish 方法CMSExpirationsend 或 publish 方法CMSPrioritysend 或 publish 方法CMSMessageIDsend 或 publis

21、h 方法CMSTimestampsend 或 publish 方法CMSCorrelationID客戶CMSReplyTo客戶CMSType客戶CMSRedeliveredCMS Provider      屬性(Properties) - 除了消息頭中定義好的標準屬性外，CMS 提供一種機制增加新屬性到消息頭中，這種新屬性包含以下幾種：    1. 應用需要用到的屬性;    2. 消息頭中原有的一些可選屬性;    3. CMS Provider 需要

22、用到的屬性。    標準的CMS 消息頭包含以下屬性：    CMSDestination -消息發送的目的地    CMSDeliveryMode -傳遞模式， 有兩種模式： PERSISTENT 和NON_PERSISTENT，PERSISTENT 表示該消息一定要被送到目的地，否則會導致應用錯誤。NON_PERSISTENT 表示偶然丟失該消息是被允許的，這兩種模式使開發者可以在消息傳遞的可靠性和吞吐量之間找到平衡點。   CMSMessageID 唯一識別每個消息的標識，由CMS Provider

23、 產生。   CMSTimestamp 一個消息被提交給CMS Provider 到消息被發出的時間。   CMSCorrelationID 用來連接到另外一個消息，典型的應用是在回復消息中連接到原消息。   CMSReplyTo 提供本消息回復消息的目的地址。   CMSRedelivered 如果一個客戶端收到一個設置了CMSRedelivered 屬性的消息，則表示可能該客戶端曾經在早些時候收到過該消息，但并沒有簽收(acknowledged)。   CMSType 消息類型的識別符。&#

24、160;  CMSExpiration 消息過期時間，等于QueueSender 的send 方法中的timeToLive 值或TopicPublisher 的publish 方法中的timeToLive 值加上發送時刻的GMT 時間值。如果timeToLive值等于零，則CMSExpiration 被設為零，表示該消息永不過期。如果發送后，在消息過期時間之后消息還沒有被發送到目的地，則該消息被清除。   CMSPriority 消息優先級，從0-9 十個級別，0-4 是普通消息，5-9 是加急消息。CMS 不要求CMS Provider 嚴格按照這十個優先級發送

25、消息，但必須保證加急消息要先于普通消息到達。    消息體(Body) - CMS API 定義了4種消息體格式，也叫消息類型，你可以使用不同形式發送接收數據并可以兼容現有的消息格式，下面描述這4種類型： 消息類型消息體TextMessagestring對象，如xml文件內容MapMessage名/值對的集合，名是string對象，值類型可以是c+任何基本類型BytesMessage字節流ObjectMessage對象類型Message沒有消息體，只有消息頭和屬性。下例演示創建并發送一個TextMessage到一個隊列：  TextMessage mes

26、sage = queueSession.createTextMessage();message.setText(msg_text); / msg_text is a Stringmessage.setCMSType(“text”);queueSender.send(message); 下例演示接收消息并轉換為合適的消息類型：  Message* m = queueReceiver.receive();If (m->getCMSType() = “text”)TextMessage txt=dynamic_cast<TextMessage*>(m);/ do

27、 somethingelse4 消息生產者客戶端消息生產者產生消息并將消息發送到broker上的隊列或主題中。要使消息生產者生產的消息被消息消費者消費，必須滿足兩個條件：生產者和消費者必須連接到同一個Broker, 即BrokerUri中主機名和端口相同生產者和消費者必須具有相同的destination, 即同一個隊列名或主題名4.1 使用activemq-cpp來創建消息生產者4.1.1 頭文件及名字空間#include <activemq/core/ActiveMQConnectionFactory.h>#include <activemq/util/Config.h&g

28、t;#include <cms/Connection.h>#include <cms/Session.h>#include <cms/TextMessage.h>#include <cms/ExceptionListener.h>#include <stdlib.h>#include <iostream>using namespace activemq;using namespace activemq:core;using namespace cms;using namespace std;4.1.2 創建一個生產者類cl

29、ass SimpleProducer private: Connection* connection; /連接對象 Session* session; /會話 Destination* destination; /消息目的地 MessageProducer* producer; /消息生產者 bool useTopic; /是否采用采用主題模式 bool clientAck; /是否自動確認消息接收 unsigned int numMessages; /生產消息數 std:string brokerURI; /連接borker uri std:string destURI; /隊列或主題名pu

30、blic:/./構造函數SimpleProducer( const std:string& brokerURI, unsigned int numMessages, const std:string& destURI, bool useTopic = false, bool clientAck = false ) connection = NULL; session = NULL; destination = NULL; producer = NULL; this->numMessages = numMessages; this->useTopic = useTop

31、ic; this->brokerURI = brokerURI; this->destURI = destURI; this->clientAck = clientAck;initialize();virtual SimpleProducer() cleanup();4.1.3 初始化及銷毀/ 初始化private:Virtual void initialize()try / 創建連接工廠 ActiveMQConnectionFactory* connectionFactory = new ActiveMQConnectionFactory( brokerURI ); / 創

32、建一個到broker的連接 connection = connectionFactory->createConnection(); connection->start(); / 關閉連接工廠 delete connectionFactory; / 創建一個會話 if( clientAck ) /消息接收后由消費者客戶端確認 session= connection->createSession( Session:CLIENT_ACKNOWLEDGE ); else /消息接收自動確認 session = connection->createSession( Session

33、:AUTO_ACKNOWLEDGE ); / 創建一個隊列或主題 (Topic or Queue) if( useTopic ) destination = session->createTopic( destURI ); else destination = session->createQueue( destURI ); / 創建生產者并設定消息傳送模式 producer = session->createProducer( destination ); producer->setDeliveryMode( DeliveryMode:NON_PERSISTENT )

34、;catch ( CMSException& e ) e.printStackTrace(); / 銷毀void cleanup() / Destroy resources. try if( destination != NULL ) delete destination; catch ( CMSException& e ) e.printStackTrace(); destination = NULL; try if( producer != NULL ) delete producer; catch ( CMSException& e ) e.printStackT

35、race(); producer = NULL; / Close open resources. try if( session != NULL ) session->close(); if( connection != NULL ) connection->close(); catch ( CMSException& e ) e.printStackTrace(); try if( session != NULL ) delete session; catch ( CMSException& e ) e.printStackTrace(); session = N

36、ULL; try if( connection != NULL ) delete connection; catch ( CMSException& e ) e.printStackTrace(); connection = NULL; 4.1.4 生產一個消息并發送到隊列中public :void send() / 消息內容 string text = (string)"Hello world! thread " for( std:size_t ix=0; ix<numMessages; +ix )/ 創建一個文本類型的消息 TextMessage* mes

37、sage = session->createTextMessage( text ); / 發送消息 printf( "Sent message #%d n", ix+1 ); producer->send( message ); / 釋放消息 delete message;4.1.5 發送消息主程序Int main(void) / broker uristd:string brokerURI = "tcp:/127.0.0.1:61616" "?wireFormat=openwire" "&transpo

38、rt.useAsyncSend=true"/ 發送消息數unsigned int numMessages = 2000;/ 消息隊列名std:string destURI = "TEST.FOO"/ 使用隊列模式bool useTopics = false; /初始化一個消息生產者對象并發送消息 SimpleProducer producer( brokerURI, numMessages, destURI, useTopics );producer.send();return 0;4.1.6 總結綜上例子可知，每次發送一個消息到消息隊列中都需要定義一個生產者類，

39、并完成一個機械的初始化及銷毀過程。為了提高軟件開發效率，可以模仿生產者類定義一個消息發送者類，封裝相關細節，并編譯成共享庫以供使用，簡化編程過程。4.2 使用winkeemq-cpp來創建消息生產者4.2.1 頭文件及名字空間#include <MessageSender.h>using namespace winkeemq;using namespace std;4.2.2 發送消息主程序int main(int argc, char* argv)/ broker uri std:string brokerURI = "tcp:/192.168.1.179:61616&

40、quot; "?wireFormat=openwire" "&wireFormat.maxInactivityDuration=0" "&soKeepAlive=true" "&transport.useAsyncSend=true"/ 隊列名string mqName="mm.mq"/ 創建一個消息發送對象(采用隊列模式，每次只發一個消息)MessageSender ms(brokerURI,1,false,mqName);string body=”hello worl

41、dn”;/ 創建一個文本消息TextMessage* msg=dynamic_cast<TextMessage* > (ms.createMessag(MessageSender:TEXT_MESSAGE);/ 設定消息體內容 msg->setText(body); / 發送消息 ms.sendMessage();/ 銷毀消息 ms.deleteMessage();4.2.3 總結由上述例子可看出，采用winkeemq-cpp后代碼量精簡了很多，開發員不需要關心那些機械的初始化細節。要創建一個消息生產者，只需要給定Broker uri, 隊列名，消息目的模式，然后調用Mess

42、ageSender的createMessage()創建一個具體類型的消息，createMessage()的參數是一個在MessageSender中定義的一個無名enum， 指明消息的類型。調用MessageSender的sendMessage()發送消息到broker中，最后銷毀消息5 消息消費者客戶端消息消費者從Broker上的隊列或主題中取出消息并做相應的處理。5.1 使用activemq-cpp來創建消息消息者5.1.1 頭文件及名字空間#include <activemq/concurrent/Thread.h>#include <activemq/concurren

43、t/Runnable.h>#include <activemq/concurrent/CountDownLatch.h>#include <activemq/core/ActiveMQConnectionFactory.h>#include <cms/Connection.h>#include <cms/Session.h>#include <cms/TextMessage.h>#include <cms/ExceptionListener.h>#include <cms/MessageListener.h&g

44、t;#include <stdlib.h>#include <iostream>using namespace activemq;using namespace activemq:core;using namespace cms;using namespace std;5.1.2 創建一個生產者類class SimpleAsyncConsumer : public ExceptionListener, public MessageListener private: Connection* connection; /連接對象 Session* session; /會話 D

45、estination* destination; /消息目的地 MessageConsumer* consumer; /消息消費者 bool useTopic; /是否采用采用主題模式 bool clientAck; /是否自動確認消息接收 std:string brokerURI; /連接borker uri std:string destURI; /隊列或主題名public:/./構造函數 SimpleAsyncConsumer( const std:string& brokerURI, const std:string& destURI, bool useTopic =

46、false, bool clientAck = false ) connection = NULL; session = NULL; destination = NULL; consumer = NULL; this->useTopic = useTopic; this->brokerURI = brokerURI; this->destURI = destURI;this->clientAck = clientAck;initialize (); virtual SimpleAsyncConsumer () cleanup();5.1.3 初始化及銷毀private:

47、/ 初始化virtual void initialize()try / 創建連接工廠 ActiveMQConnectionFactory* connectionFactory = new ActiveMQConnectionFactory( brokerURI ); / 創建一個到broker的連接 connection = connectionFactory->createConnection(); connection->start(); / 設置連接異常偵聽類connection->setExceptionListener(this); / 關閉連接工廠 delete

48、connectionFactory; / 創建一個會話 if( clientAck ) /消息接收后由消費者客戶端確認 session= connection->createSession( Session:CLIENT_ACKNOWLEDGE ); else /消息接收自動確認 session = connection->createSession( Session:AUTO_ACKNOWLEDGE ); / 創建一個隊列或主題 (Topic or Queue) if( useTopic ) destination = session->createTopic( destU

49、RI ); else destination = session->createQueue( destURI ); / 創建消費者并設定消息接收偵聽類 consumer = session->createConsumer( destination ); consumer->setMessageListener( this );catch ( CMSException& e ) e.printStackTrace(); / 銷毀void cleanup() / Destroy resources. try if( destination != NULL ) delete

50、 destination; catch ( CMSException& e ) e.printStackTrace(); destination = NULL; try if( producer != NULL ) delete producer; catch ( CMSException& e ) e.printStackTrace(); producer = NULL; / Close open resources. try if( session != NULL ) session->close(); if( connection != NULL ) connect

51、ion->close(); catch ( CMSException& e ) e.printStackTrace(); try if( session != NULL ) delete session; catch ( CMSException& e ) e.printStackTrace(); session = NULL; try if( connection != NULL ) delete connection; catch ( CMSException& e ) e.printStackTrace(); connection = NULL; 5.1.4

52、 從消息隊列中異步接收消息如果隊列中有消息到來，程序會自動調用onMessage函數，因此只需要在onMessage()中編寫對消息的處理。onMessage()函數中的message參數在onMessage()返回后便會自動銷毀，可以通過調用Message的clone()方法拷貝自身來擴展其生命周期，clone()返回的是在椎上分配的消息，因此需要手動銷毀。virtual void onMessage( const Message* message ) static int count = 0; try count+; / 強制轉換消息到實際所屬類型 const TextMessage* textMessage =