\u5b9e\u9645\u5e94\u7528\u5c31\u662f\u8ba9\u5f00\u53d1\u8005\u80fd\u4ece\u591a\u7ebf\u7a0b,\u591a\u6d88\u606f\u901a\u4fe1\u4e2d\u89e3\u6551\u51fa\u6765.\u66f4\u591a\u7684\u5173\u6ce8\u5e94\u7528\u903b\u8f91.<\/p>\n
CMS (stands for C++ Messaging Service) is a JMS-like API for C++ for interfacing with Message Brokers such as\u00a0Apache ActiveMQ. CMS helps to make your C++ client code much neater and easier to follow. To get a better feel for CMS try the\u00a0API\u00a0Reference. ActiveMQ-CPP is a client only library, a message broker such as\u00a0Apache ActiveMQ\u00a0is still needed for your clients to communicate.<\/p>\n
Our implementation of CMS is called ActiveMQ-CPP, which has an architecture that allows for pluggable transports and wire formats. Currently we support the\u00a0OpenWire\u00a0and\u00a0Stomp\u00a0protocols, both over TCP and SSL, we also now support a Failover Transport for more reliable client operation. In addition to CMS, ActiveMQ-CPP also provides a robust set of classes that support platform independent constructs such as threading, I\/O, sockets, etc. You may find many of these utilities very useful, such as a Java like Thread class or the “synchronized” macro that let’s you use a Java-like synchronization on any object that implements the activemq::concurrent::Synchronizable interface. ActiveMQ-CPP is released under the\u00a0Apache\u00a02.0 License<\/p>\n
CMS (C++ \u6d88\u606f \u670d\u52a1)\u662f\u4e00\u4e2a\u9762\u8c61apache activemq \u7684 \u6d88\u606f \u4e2d\u95f4\u5c42\u7684C++\u63a5\u53e3.<\/p>\n
CMS\u7684\u5b9e\u73b0 \u53eb\u505aactivemq-cpp ,\u4e0d\u8fc7\u5f53\u524d\u53ea\u652f\u6301\u00a0openwire,amqp,TCP,ssl. \u73b0\u5728\u8fd8\u652f\u6301 \u4e3b\u5907\u5207\u6362\u529f\u80fd(\u8fd9\u4e2a\u662f\u91cd\u70b9,\u5f53\u65f6\u6211\u4e0d\u61c2,\u7ed3\u679c\u5c31\u8d70\u4e86\u5f2f\u8def!_!).<\/p>\n
-_- ,\u610f\u601d\u662f activemq\\conf\\activemq.xml\u4e2d\u7684stomp,mqtt,ws \u662f\u6ca1\u529e\u6cd5\u7684.<\/p>\n
$\"\u590d\u5236\u4ee3\u7801\"$ <\/span><\/div>\n
<transportConnectors>\r\n<!-- DOS protection, limit concurrent connections to 1000 and frame size to 100MB -->\r\n<transportConnector name=\"openwire\" uri=\"tcp:\/\/0.0.0.0:61616?maximumConnections=1000&amp;wireFormat.maxFrameSize=104857600\"\/>\r\n<transportConnector name=\"amqp\" uri=\"amqp:\/\/0.0.0.0:5672?maximumConnections=1000&amp;wireFormat.maxFrameSize=104857600\"\/>\r\n<transportConnector name=\"stomp\" uri=\"stomp:\/\/0.0.0.0:61613?maximumConnections=1000&amp;wireFormat.maxFrameSize=104857600\"\/>\r\n<transportConnector name=\"mqtt\" uri=\"mqtt:\/\/0.0.0.0:1883?maximumConnections=1000&amp;wireFormat.maxFrameSize=104857600\"\/>\r\n<transportConnector name=\"ws\" uri=\"ws:\/\/0.0.0.0:61614?maximumConnections=1000&amp;wireFormat.maxFrameSize=104857600\"\/>\r\n<\/transportConnectors><\/pre>\n<\/span><\/div>\n<\/div>\n <\/p>\n
 <\/p>\n
 <\/p>\n
1,acticvemq-cpp \u7684\u914d\u7f6e\u4f7f\u7528.<\/h1>\n\u53c2\u8003:Active MQ \u00a0C++\u5b9e\u73b0\u901a\u8baf\u00a0http:\/\/blog.csdn.net\/lee353086\/article\/details\/6777261<\/p>\n
 <\/p>\n
activemq-cpp\u4e0b\u8f7d\u5730\u5740:<\/p>\n
http:\/\/activemq.apache.org\/cms\/download.html<\/p>\n
\u76f8\u5173\u4f9d\u8d56\u5e93<\/p>\n
\u5728\u00a0http:\/\/activemq.apache.org\/cms\/building.html \u00a0\u4e2d\u662f\u6709\u4ecb\u7ecd\u7684.\u4e0d\u8fc7\u662fen\u7684.<\/p>\n
\u8fd8\u662f\u518d\u8bf4\u4e0b\u5427.\u672c\u4ebaen\u4e5f\u7279\u5dee.<\/p>\n
“With versions of ActiveMQ-CPP 2.2 and later, we have a dependency on the\u00a0Apache Portable Runtime\u00a0project. You’ll need to install APR on your system before you’ll be able to build ActiveMQ-CPP.”<\/p>\n
“The package contains a complete set of CppUnit tests. In order for you to build an run the tests, you will need to download and install the CppUnit library. See\u00a0http:\/\/cppunit.sourceforge.net\/cppunit-wiki”<\/p>\n
\u6240\u4ee5\u5c31\u5305\u542b\u4e86:apr,apr-iconv,apr-util,cppunit.<\/p>\n
 <\/p>\n
http:\/\/mirrors.hust.edu.cn\/apache\/apr\/<\/p>\n
\u4e2d\u53ef\u4ee5\u4e0b\u8f7d\u00a0apr,apr-iconv,apr-util(\u7248\u672c\u53f7\u90fd\u627e\u6700\u9ad8\u7684,\u4e0d\u8981\u4e00\u9ad8\u4e00\u4f4e,\u4e0d\u7136\u7f16\u8bd1\u4f1a\u51fa\u95ee\u9898).<\/p>\n
apr-1.5.1-win32-src.zip,<\/p>\n
apr-iconv-1.2.1-win32-src-r2.zip,<\/p>\n
apr-util-1.5.4-win32-src.zip.<\/p>\n
 <\/p>\n
\u89e3\u538b\u540e\u8bb0\u5f97\u91cd\u547d\u4ee4\u6587\u4ef6\u5939,\u53bb\u6389\u7248\u672c\u53f7,\u6539\u6210\u5982\u4e0b\u56fe,\u4e0d\u7136\u5de5\u7a0b\u7f16\u8bd1\u65f6\u9ed8\u8ba4\u7684 [\u9644\u52a0\u5305\u542b\u76ee\u5f55] \u662f\u627e\u4e0d\u5230\u7684.<\/p>\n
\u6240\u6709\u6587\u4ef6\u5939\u653e\u5728\u4e00\u4e2a\u6839\u76ee\u5f55\u4e0b.<\/p>\n
<\/p>\n
<\/p>\n
 <\/p>\n
\u6253\u5f00 \u00a0activemq-cpp-library\\vs2008-build\\activemq-cpp.sln<\/p>\n
\u4f9d\u6b21\u6dfb\u52a0[\u73b0\u5728\u9879\u76ee]:libapr.vcproj,libapriconv.vcproj,libaprutil.vcproj. \u53ea\u9700\u8981lib\u9879\u5c31\u884c\u4e86.<\/p>\n
<\/p>\n
 <\/p>\n
\u6700\u540e\u9879\u76ee\u56fe:<\/p>\n
<\/p>\n
 <\/p>\n
libapriconv.vcproj,libaprutil.vcproj \u7684[\u9879\u76ee\u4f9d\u8d56\u9879]\u90fd\u9700\u8981libapr<\/p>\n
<\/p>\n
 <\/p>\n
activemq-cpp\u00a0\u7684[\u9879\u76ee\u4f9d\u8d56\u9879]\u9700\u8981libapriconv,libaprutil,libapr.<\/p>\n
activemq-cpp \u7684[\u9644\u52a0\u5305\u542b\u76ee\u5f55] \u9700\u8981\u5305\u542b \u8fd9\u4e09\u4e2a\u7684\u7684include\u76ee\u5f55.<\/p>\n
<\/p>\n
 <\/p>\n
\u7ecf\u8fc7\u6f2b\u957f\u7684\u7f16\u8bd1\u540e,<\/p>\n
<\/p>\n
 <\/p>\n
\u8fd9\u4e2a\u5927lib\u6587\u4ef6\u5c31\u51fa\u6765.<\/p>\n
activemq-cpp-example \u8fd9\u4e2a\u5de5\u7a0b ,\u5c31\u6709 hello world \u7684\u4ee3\u7801.<\/p>\n
 <\/p>\n
2,activemq-cpp-example \u9879\u76ee\u4ee3\u7801\u89e3\u6790.<\/h1>\n\u901a\u8fc7\u8fd9\u4e2a\u9879\u76ee\u53ef\u4ee5\u8ba9\u6211\u4eec\u66f4\u597d\u7684\u8ba4\u8bc6 activemq-cpp\u7684\u7ed3\u6784.<\/p>\n
\n<\/span><\/div>\n\n<\/span><\/div>\n\/*\r\n * Licensed to the Apache Software Foundation (ASF) under one or more\r\n * contributor license agreements.  See the NOTICE file distributed with\r\n * this work for additional information regarding copyright ownership.\r\n * The ASF licenses this file to You under the Apache License, Version 2.0\r\n * (the \"License\"); you may not use this file except in compliance with\r\n * the License.  You may obtain a copy of the License at\r\n *\r\n *     http:\/\/www.apache.org\/licenses\/LICENSE-2.0\r\n *\r\n * Unless required by applicable law or agreed to in writing, software\r\n * distributed under the License is distributed on an \"AS IS\" BASIS,\r\n * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\r\n * See the License for the specific language governing permissions and\r\n * limitations under the License.\r\n *\/\r\n\r\n\/\/ START SNIPPET: demo\r\n\r\n#include <activemq\/library\/ActiveMQCPP.h>\r\n#include <decaf\/lang\/Thread.h>\r\n#include <decaf\/lang\/Runnable.h>\r\n#include <decaf\/util\/concurrent\/CountDownLatch.h>\r\n#include <decaf\/lang\/Integer.h>\r\n#include <decaf\/lang\/Long.h>\r\n#include <decaf\/lang\/System.h>\r\n#include <activemq\/core\/ActiveMQConnectionFactory.h>\r\n#include <activemq\/util\/Config.h>\r\n#include <cms\/Connection.h>\r\n#include <cms\/Session.h>\r\n#include <cms\/TextMessage.h>\r\n#include <cms\/BytesMessage.h>\r\n#include <cms\/MapMessage.h>\r\n#include <cms\/ExceptionListener.h>\r\n#include <cms\/MessageListener.h>\r\n#include <stdlib.h>\r\n#include <stdio.h>\r\n#include <iostream>\r\n#include <memory>\r\n\r\nusing namespace activemq::core;\r\nusing namespace decaf::util::concurrent;\r\nusing namespace decaf::util;\r\nusing namespace decaf::lang;\r\nusing namespace cms;\r\nusing namespace std;\r\n\r\nclass HelloWorldProducer : public Runnable {\r\nprivate:\r\n\r\n    Connection* connection;\r\n    Session* session;\r\n    Destination* destination;\r\n    MessageProducer* producer;\r\n    int numMessages;\r\n    bool useTopic;\r\n    bool sessionTransacted;\r\n    std::string brokerURI;\r\n\r\nprivate:\r\n\r\n    HelloWorldProducer(const HelloWorldProducer&);\r\n    HelloWorldProducer& operator=(const HelloWorldProducer&);\r\n\r\npublic:\r\n\r\n    HelloWorldProducer(const std::string& brokerURI, int numMessages, bool useTopic = false, bool sessionTransacted = false) :\r\n        connection(NULL),\r\n        session(NULL),\r\n        destination(NULL),\r\n        producer(NULL),\r\n        numMessages(numMessages),\r\n        useTopic(useTopic),\r\n        sessionTransacted(sessionTransacted),\r\n        brokerURI(brokerURI) {\r\n    }\r\n\r\n    virtual ~HelloWorldProducer(){\r\n        cleanup();\r\n    }\r\n\r\n    void close() {\r\n        this->cleanup();\r\n    }\r\n\r\n    virtual void run() {\r\n\r\n        try {\r\n\r\n            \/\/ Create a ConnectionFactory\r\n            auto_ptr<ConnectionFactory> connectionFactory(\r\n                ConnectionFactory::createCMSConnectionFactory(brokerURI));\r\n\r\n            \/\/ Create a Connection\r\n            connection = connectionFactory->createConnection();\r\n            connection->start();\r\n\r\n            \/\/ Create a Session\r\n            if (this->sessionTransacted) {\r\n                session = connection->createSession(Session::SESSION_TRANSACTED);\r\n            } else {\r\n                session = connection->createSession(Session::AUTO_ACKNOWLEDGE);\r\n            }\r\n\r\n            \/\/ Create the destination (Topic or Queue)\r\n            if (useTopic) {\r\n                destination = session->createTopic(\"TEST.FOO\");\r\n            } else {\r\n                destination = session->createQueue(\"TEST.FOO\");\r\n            }\r\n\r\n            \/\/ Create a MessageProducer from the Session to the Topic or Queue\r\n            producer = session->createProducer(destination);\r\n            producer->setDeliveryMode(DeliveryMode::NON_PERSISTENT);\r\n\r\n            \/\/ Create the Thread Id String\r\n            string threadIdStr = Long::toString(Thread::currentThread()->getId());\r\n\r\n            \/\/ Create a messages\r\n            string text = (string) \"Hello world! from thread \" + threadIdStr;\r\n\r\n            for (int ix = 0; ix < numMessages; ++ix) {\r\n                std::auto_ptr<TextMessage> message(session->createTextMessage(text));\r\n                message->setIntProperty(\"Integer\", ix);\r\n                printf(\"Sent message #%d from thread %s\\n\", ix + 1, threadIdStr.c_str());\r\n                producer->send(message.get());\r\n            }\r\n\r\n        } catch (CMSException& e) {\r\n            e.printStackTrace();\r\n        }\r\n    }\r\n\r\nprivate:\r\n\r\n    void cleanup() {\r\n\r\n        if (connection != NULL) {\r\n            try {\r\n                connection->close();\r\n            } catch (cms::CMSException& ex) {\r\n                ex.printStackTrace();\r\n            }\r\n        }\r\n\r\n        \/\/ Destroy resources.\r\n        try {\r\n            delete destination;\r\n            destination = NULL;\r\n            delete producer;\r\n            producer = NULL;\r\n            delete session;\r\n            session = NULL;\r\n            delete connection;\r\n            connection = NULL;\r\n        } catch (CMSException& e) {\r\n            e.printStackTrace();\r\n        }\r\n    }\r\n};\r\n\r\nclass HelloWorldConsumer : public ExceptionListener,\r\n                           public MessageListener,\r\n                           public Runnable {\r\n\r\nprivate:\r\n\r\n    CountDownLatch latch;\r\n    CountDownLatch doneLatch;\r\n    Connection* connection;\r\n    Session* session;\r\n    Destination* destination;\r\n    MessageConsumer* consumer;\r\n    long waitMillis;\r\n    bool useTopic;\r\n    bool sessionTransacted;\r\n    std::string brokerURI;\r\n\r\nprivate:\r\n\r\n    HelloWorldConsumer(const HelloWorldConsumer&);\r\n    HelloWorldConsumer& operator=(const HelloWorldConsumer&);\r\n\r\npublic:\r\n\r\n    HelloWorldConsumer(const std::string& brokerURI, int numMessages, bool useTopic = false, bool sessionTransacted = false, int waitMillis = 30000) :\r\n        latch(1),\r\n        doneLatch(numMessages),\r\n        connection(NULL),\r\n        session(NULL),\r\n        destination(NULL),\r\n        consumer(NULL),\r\n        waitMillis(waitMillis),\r\n        useTopic(useTopic),\r\n        sessionTransacted(sessionTransacted),\r\n        brokerURI(brokerURI) {\r\n    }\r\n\r\n    virtual ~HelloWorldConsumer() {\r\n        cleanup();\r\n    }\r\n\r\n    void close() {\r\n        this->cleanup();\r\n    }\r\n\r\n    void waitUntilReady() {\r\n        latch.await();\r\n    }\r\n\r\n    virtual void run() {\r\n\r\n        try {\r\n\r\n            \/\/ Create a ConnectionFactory\r\n            auto_ptr<ConnectionFactory> connectionFactory(\r\n                ConnectionFactory::createCMSConnectionFactory(brokerURI));\r\n\r\n            \/\/ Create a Connection\r\n            connection = connectionFactory->createConnection();\r\n            connection->start();\r\n            connection->setExceptionListener(this);\r\n\r\n            \/\/ Create a Session\r\n            if (this->sessionTransacted == true) {\r\n                session = connection->createSession(Session::SESSION_TRANSACTED);\r\n            } else {\r\n                session = connection->createSession(Session::AUTO_ACKNOWLEDGE);\r\n            }\r\n\r\n            \/\/ Create the destination (Topic or Queue)\r\n            if (useTopic) {\r\n                destination = session->createTopic(\"TEST.FOO\");\r\n            } else {\r\n                destination = session->createQueue(\"TEST.FOO\");\r\n            }\r\n\r\n            \/\/ Create a MessageConsumer from the Session to the Topic or Queue\r\n            consumer = session->createConsumer(destination);\r\n\r\n            consumer->setMessageListener(this);\r\n\r\n            std::cout.flush();\r\n            std::cerr.flush();\r\n\r\n            \/\/ Indicate we are ready for messages.\r\n            latch.countDown();\r\n\r\n            \/\/ Wait while asynchronous messages come in.\r\n            doneLatch.await(waitMillis);\r\n\r\n        } catch (CMSException& e) {\r\n            \/\/ Indicate we are ready for messages.\r\n            latch.countDown();\r\n            e.printStackTrace();\r\n        }\r\n    }\r\n\r\n    \/\/ Called from the consumer since this class is a registered MessageListener.\r\n    virtual void onMessage(const Message* message) {\r\n\r\n        static int count = 0;\r\n\r\n        try {\r\n            count++;\r\n            const TextMessage* textMessage = dynamic_cast<const TextMessage*> (message);\r\n            string text = \"\";\r\n\r\n            if (textMessage != NULL) {\r\n                text = textMessage->getText();\r\n            } else {\r\n                text = \"NOT A TEXTMESSAGE!\";\r\n            }\r\n\r\n            printf(\"Message #%d Received: %s\\n\", count, text.c_str());\r\n\r\n        } catch (CMSException& e) {\r\n            e.printStackTrace();\r\n        }\r\n\r\n        \/\/ Commit all messages.\r\n        if (this->sessionTransacted) {\r\n            session->commit();\r\n        }\r\n\r\n        \/\/ No matter what, tag the count down latch until done.\r\n        doneLatch.countDown();\r\n    }\r\n\r\n    \/\/ If something bad happens you see it here as this class is also been\r\n    \/\/ registered as an ExceptionListener with the connection.\r\n    virtual void onException(const CMSException& ex AMQCPP_UNUSED) {\r\n        printf(\"CMS Exception occurred.  Shutting down client.\\n\");\r\n        ex.printStackTrace();\r\n        exit(1);\r\n    }\r\n\r\nprivate:\r\n\r\n    void cleanup() {\r\n        if (connection != NULL) {\r\n            try {\r\n                connection->close();\r\n            } catch (cms::CMSException& ex) {\r\n                ex.printStackTrace();\r\n            }\r\n        }\r\n\r\n        \/\/ Destroy resources.\r\n        try {\r\n            delete destination;\r\n            destination = NULL;\r\n            delete consumer;\r\n            consumer = NULL;\r\n            delete session;\r\n            session = NULL;\r\n            delete connection;\r\n            connection = NULL;\r\n        } catch (CMSException& e) {\r\n            e.printStackTrace();\r\n        }\r\n    }\r\n};\r\n\r\nint main(int argc AMQCPP_UNUSED, char* argv[] AMQCPP_UNUSED) {\r\n\r\n    activemq::library::ActiveMQCPP::initializeLibrary();\r\n    {\r\n    std::cout << \"=====================================================\\n\";\r\n    std::cout << \"Starting the example:\" << std::endl;\r\n    std::cout << \"-----------------------------------------------------\\n\";\r\n\r\n\r\n    \/\/ Set the URI to point to the IP Address of your broker.\r\n    \/\/ add any optional params to the url to enable things like\r\n    \/\/ tightMarshalling or tcp logging etc.  See the CMS web site for\r\n    \/\/ a full list of configuration options.\r\n    \/\/\r\n    \/\/  http:\/\/activemq.apache.org\/cms\/\r\n    \/\/\r\n    \/\/ Wire Format Options:\r\n    \/\/ =========================\r\n    \/\/ Use either stomp or openwire, the default ports are different for each\r\n    \/\/\r\n    \/\/ Examples:\r\n    \/\/    tcp:\/\/127.0.0.1:61616                      default to openwire\r\n    \/\/    tcp:\/\/127.0.0.1:61616?wireFormat=openwire  same as above\r\n    \/\/    tcp:\/\/127.0.0.1:61613?wireFormat=stomp     use stomp instead\r\n    \/\/\r\n    \/\/ SSL:\r\n    \/\/ =========================\r\n    \/\/ To use SSL you need to specify the location of the trusted Root CA or the\r\n    \/\/ certificate for the broker you want to connect to.  Using the Root CA allows\r\n    \/\/ you to use failover with multiple servers all using certificates signed by\r\n    \/\/ the trusted root.  If using client authentication you also need to specify\r\n    \/\/ the location of the client Certificate.\r\n    \/\/\r\n    \/\/     System::setProperty( \"decaf.net.ssl.keyStore\", \"<path>\/client.pem\" );\r\n    \/\/     System::setProperty( \"decaf.net.ssl.keyStorePassword\", \"password\" );\r\n    \/\/     System::setProperty( \"decaf.net.ssl.trustStore\", \"<path>\/rootCA.pem\" );\r\n    \/\/\r\n    \/\/ The you just specify the ssl transport in the URI, for example:\r\n    \/\/\r\n    \/\/     ssl:\/\/localhost:61617\r\n    \/\/\r\n    std::string brokerURI =\r\n        \"failover:(tcp:\/\/localhost:61616\"\r\n\/\/        \"?wireFormat=openwire\"\r\n\/\/        \"&transport.useInactivityMonitor=false\"\r\n\/\/        \"&connection.alwaysSyncSend=true\"\r\n\/\/        \"&connection.useAsyncSend=true\"\r\n\/\/        \"?transport.commandTracingEnabled=true\"\r\n\/\/        \"&transport.tcpTracingEnabled=true\"\r\n\/\/        \"&wireFormat.tightEncodingEnabled=true\"\r\n        \")\";\r\n\r\n    \/\/============================================================\r\n    \/\/ set to true to use topics instead of queues\r\n    \/\/ Note in the code above that this causes createTopic or\r\n    \/\/ createQueue to be used in both consumer an producer.\r\n    \/\/============================================================\r\n    bool useTopics = true;\r\n    bool sessionTransacted = false;\r\n    int numMessages = 2000;\r\n\r\n    long long startTime = System::currentTimeMillis();\r\n\r\n    HelloWorldProducer producer(brokerURI, numMessages, useTopics);\r\n        HelloWorldConsumer consumer(brokerURI, numMessages, useTopics, sessionTransacted);\r\n\r\n    \/\/ Start the consumer thread.\r\n    Thread consumerThread(&consumer);\r\n    consumerThread.start();\r\n\r\n    \/\/ Wait for the consumer to indicate that its ready to go.\r\n    consumer.waitUntilReady();\r\n\r\n    \/\/ Start the producer thread.\r\n    Thread producerThread(&producer);\r\n    producerThread.start();\r\n\r\n    \/\/ Wait for the threads to complete.\r\n    producerThread.join();\r\n    consumerThread.join();\r\n\r\n    long long endTime = System::currentTimeMillis();\r\n    double totalTime = (double)(endTime - startTime) \/ 1000.0;\r\n\r\n    consumer.close();\r\n    producer.close();\r\n\r\n    std::cout << \"Time to completion = \" << totalTime << \" seconds.\" << std::endl;\r\n    std::cout << \"-----------------------------------------------------\\n\";\r\n    std::cout << \"Finished with the example.\" << std::endl;\r\n    std::cout << \"=====================================================\\n\";\r\n\r\n    }\r\n    activemq::library::ActiveMQCPP::shutdownLibrary();\r\n}\r\n\r\n\/\/ END SNIPPET: demo<\/pre>\n<\/span><\/div>\n<\/div>\n <\/p>\n
<\/span><\/div>\n<\/div>\n\u4ecemain()\u5f00\u59cb\u5427.<\/p>\n
<\/p>\n
 <\/p>\n
<\/p>\n
 <\/p>\n
 <\/p>\n
\u8fd9\u6837\u4fbf\u7b80\u5feb\u901f\u7684\u5b9e\u73b0\u4e86\u5e94\u7528\u903b\u8f91.<\/p>\n
 <\/p>\n
 <\/p>\n
3,activemq\u7684\u51e0\u79cd\u901a\u4fe1\u6a21\u5f0f.<\/h1>\n\u53ef\u4ee5\u53c2\u8003:<\/p>\n
http:\/\/shmilyaw-hotmail-com.iteye.com\/blog\/1897635<\/p>\n
 <\/p>\n
\u76ee\u524d\u672c\u4eba\u9700\u8981\u7684\u662factivemq-cpp\u7684\u00a0request-response \u6a21\u5f0f.<\/p>\n
 <\/p>\n
 <\/p>\n
 <\/p>\n
4,activemq-cpp\u7684\u00a0request-response \u6a21\u5f0f\u7684\u5e94\u7528.<\/h1>\n\u670d\u52a1\u5668\u4e0e\u5ba2\u6237\u7aef\u901a\u4fe1 \u6570\u636e\u7684\u4ea4\u4e92 \u548c \u786e\u8ba4.<\/p>\n
<\/p>\n
 <\/p>\n
\u4ee5\u4e0b\u662f\u672c\u4eba\u4fee\u6539\u540e\u7684\u7b80\u5355\u4ee3\u7801,bug\u53ef\u80fd\u5b58\u5728,\u8bf7\u6307\u51fa.<\/p>\n
 <\/p>\n
\u590d\u5236\u4e24\u4efd,\u4e00\u4efd\u5b9a\u4e49\u00a0USE_COMSUMER \u4e00\u4efd\u5b9a\u4e49\u00a0USE_PRODUCER \u5c31\u53ef\u4ee5\u751f\u6210.<\/p>\n
\n<\/span><\/div>\n\/*\r\n * Licensed to the Apache Software Foundation (ASF) under one or more\r\n * contributor license agreements.  See the NOTICE file distributed with\r\n * this work for additional information regarding copyright ownership.\r\n * The ASF licenses this file to You under the Apache License, Version 2.0\r\n * (the \"License\"); you may not use this file except in compliance with\r\n * the License.  You may obtain a copy of the License at\r\n *\r\n *     http:\/\/www.apache.org\/licenses\/LICENSE-2.0\r\n *\r\n * Unless required by applicable law or agreed to in writing, software\r\n * distributed under the License is distributed on an \"AS IS\" BASIS,\r\n * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\r\n * See the License for the specific language governing permissions and\r\n * limitations under the License.\r\n *\/\r\n\r\n\/\/ START SNIPPET: demo\r\n\r\n#include <activemq\/library\/ActiveMQCPP.h>\r\n#include <decaf\/lang\/Thread.h>\r\n#include <decaf\/lang\/Runnable.h>\r\n#include <decaf\/util\/concurrent\/CountDownLatch.h>\r\n#include <decaf\/lang\/Integer.h>\r\n#include <decaf\/lang\/Long.h>\r\n#include <decaf\/lang\/System.h>\r\n#include <activemq\/core\/ActiveMQConnectionFactory.h>\r\n#include <activemq\/util\/Config.h>\r\n#include <cms\/Connection.h>\r\n#include <cms\/Session.h>\r\n#include <cms\/TextMessage.h>\r\n#include <cms\/BytesMessage.h>\r\n#include <cms\/MapMessage.h>\r\n#include <cms\/ExceptionListener.h>\r\n#include <cms\/MessageListener.h>\r\n#include <stdlib.h>\r\n#include <stdio.h>\r\n#include <iostream>\r\n#include <memory>\r\n\r\n#include <decaf\/util\/Random.h>\r\n\r\nusing namespace activemq::core;\r\nusing namespace decaf::util::concurrent;\r\nusing namespace decaf::util;\r\nusing namespace decaf::lang;\r\nusing namespace cms;\r\nusing namespace std;\r\n\r\n#define  QUEUE_NAME    \"eventQueue\"\r\n#define NAME_BYTE_LEN        16\r\n\r\nclass HelloWorldProducer : public ExceptionListener,\r\n                            public MessageListener,\r\n                            public Runnable {\r\nprivate:\r\n    CountDownLatch latch;\r\n    CountDownLatch doneLatch;\r\n    Connection* connection;\r\n    Session* session;\r\n    Destination* destination;\r\n    MessageProducer* producer;\r\n    int numMessages;\r\n    bool useTopic;\r\n    bool sessionTransacted;\r\n    std::string brokerURI;\r\n    bool bReciveMessage;\r\n    long waitMillis;\r\n\r\nprivate:\r\n\r\n    HelloWorldProducer(const HelloWorldProducer&);\r\n    HelloWorldProducer& operator=(const HelloWorldProducer&);\r\n\r\npublic:\r\n\r\n    HelloWorldProducer(const std::string& brokerURI, int numMessages, bool useTopic = false, bool sessionTransacted = false,\r\n        long waitMillis=3000) :\r\n        latch(1),\r\n        doneLatch(numMessages),  \r\n        connection(NULL),\r\n        session(NULL),\r\n        destination(NULL),\r\n        producer(NULL),\r\n        numMessages(numMessages),\r\n        useTopic(useTopic),\r\n        sessionTransacted(sessionTransacted),\r\n        brokerURI(brokerURI) ,\r\n        bReciveMessage(false),\r\n        waitMillis(waitMillis)\r\n        { }\r\n\r\n    virtual ~HelloWorldProducer(){\r\n        cleanup();\r\n    }\r\n\r\n    void close() {\r\n        this->cleanup();\r\n    }\r\n    \r\n    void waitUntilReady() {\r\n        latch.await();\r\n    }\r\n\r\n    virtual void run() {\r\n\r\n        try {\r\n\r\n            \/\/ Create a ConnectionFactory\r\n            auto_ptr<ConnectionFactory> connectionFactory(\r\n                ConnectionFactory::createCMSConnectionFactory(brokerURI));\r\n\r\n            \/\/ Create a Connection\r\n            connection = connectionFactory->createConnection();\r\n            connection->start();\r\n\r\n            \/\/ Create a Session\r\n            if (this->sessionTransacted) {\r\n                session = connection->createSession(Session::SESSION_TRANSACTED);\r\n            } else {\r\n                session = connection->createSession(Session::AUTO_ACKNOWLEDGE);\r\n            }\r\n            \r\n            session = connection->createSession();\r\n            \/\/ Create the destination (Topic or Queue)\r\n            if (useTopic) {\r\n                destination = session->createTopic(QUEUE_NAME);\r\n            } else {\r\n                destination = session->createQueue(QUEUE_NAME);\r\n            }\r\n\r\n            \/\/ Create a MessageProducer from the Session to the Topic or Queue\r\n            producer = session->createProducer(destination);\r\n            producer->setDeliveryMode(DeliveryMode::NON_PERSISTENT);\r\n\r\n            \/\/ Create the Thread Id String\r\n            string threadIdStr = Long::toString(Thread::currentThread()->getId());\r\n\r\n            \/\/ Create a messages\r\n            string text = (string) \"Hello world! from thread \" + threadIdStr;\r\n\r\n            for (int ix = 0; ix < numMessages; ++ix) {\r\n                std::auto_ptr<TextMessage> message(session->createTextMessage(text));\r\n                \r\n                \/\/\u5173\u952e\u6d88\u606f...\r\n                std::auto_ptr<Destination> tempDest(session->createTemporaryQueue());\r\n\r\n                \/\/cms::Destination tempDest=session->createTemporaryTopic() ; \r\n                MessageConsumer * responseConsumer = session->createConsumer(tempDest.get());  \r\n                responseConsumer->setMessageListener(this);\/\/\u76d1\u542c...\r\n\r\n\r\n                message->setCMSReplyTo(tempDest.get());\r\n                Random random;\r\n                char buffer[NAME_BYTE_LEN]={0};\r\n                random.nextBytes((unsigned char *)buffer,NAME_BYTE_LEN);\r\n                string correlationId=\"\";\r\n                for(int i=0;i<NAME_BYTE_LEN;++i)\r\n                {\r\n                    char ch[NAME_BYTE_LEN*2]={0};\r\n                    sprintf(ch,\"%02X\",(unsigned char)buffer[i]);\r\n                    string str(ch);\r\n\r\n                    correlationId+=str;\r\n                }\r\n\r\n                message->setCMSCorrelationID(correlationId);\r\n\r\n                message->setIntProperty(\"Integer\", ix);\r\n                printf(\"Producer Sent message #%d from thread %s\\n\", ix + 1, threadIdStr.c_str());\r\n                producer->send(message.get());\r\n\r\n                \/\/ Indicate we are ready for messages.\r\n                latch.countDown();\r\n\r\n                \/\/ Wait while asynchronous messages come in.\r\n                doneLatch.await(waitMillis);\r\n\r\n            } \r\n        }    \r\n        catch (CMSException& e) {\r\n            printf(\"Producer run() CMSException \\n\" );\r\n                \/\/ Indicate we are ready for messages.\r\n                latch.countDown();\r\n                e.printStackTrace();\r\n            }\r\n\r\n    \r\n        }\r\n        \r\n\r\n    \/\/ Called from the Producer since this class is a registered MessageListener.\r\n    virtual void onMessage(const Message* message) {\r\n\r\n        static int count = 0;\r\n\r\n        try {\r\n            count++;\r\n            const TextMessage* textMessage = dynamic_cast<const TextMessage*> (message);\r\n            \/\/ActiveMQMessageTransformation\r\n            \/\/std::auto_ptr<TextMessage> responsemessage(session->createTextMessage());\r\n            \/\/responsemessage->setCMSCorrelationID(textMessage->getCMSCorrelationID());\r\n            \/\/responsemessage->getCMSReplyTo()\r\n\r\n            string text = \"\";\r\n\r\n            if (textMessage != NULL) {\r\n                text = textMessage->getText();\r\n            } else {\r\n                text = \"NOT A TEXTMESSAGE!\";\r\n            }\r\n\r\n            printf(\"Producer Message #%d Received: %s\\n\", count, text.c_str());\r\n\r\n\r\n            \/\/producer.send\r\n\r\n        } catch (CMSException& e) {\r\n            printf(\"Producer onMessage() CMSException \\n\" );\r\n            e.printStackTrace();\r\n        }\r\n\r\n        \/\/ Commit all messages.\r\n        if (this->sessionTransacted) {\r\n            session->commit();\r\n        }\r\n\r\n        \/\/ No matter what, tag the count down latch until done.\r\n        doneLatch.countDown();\r\n    }\r\n\r\n    \/\/ If something bad happens you see it here as this class is also been\r\n    \/\/ registered as an ExceptionListener with the connection.\r\n    virtual void onException(const CMSException& ex AMQCPP_UNUSED) {\r\n        printf(\"Producer onException() CMS Exception occurred.  Shutting down client. \\n\" );\r\n        ex.printStackTrace();\r\n        exit(1);\r\n    }\r\n\r\n\r\nprivate:\r\n\r\n    void cleanup() {\r\n\r\n        if (connection != NULL) {\r\n            try {\r\n                connection->close();\r\n            } catch (cms::CMSException& ex) {\r\n                ex.printStackTrace();\r\n            }\r\n        }\r\n\r\n        \/\/ Destroy resources.\r\n        try {\r\n            delete destination;\r\n            destination = NULL;\r\n            delete producer;\r\n            producer = NULL;\r\n            delete session;\r\n            session = NULL;\r\n            delete connection;\r\n            connection = NULL;\r\n        } catch (CMSException& e) {\r\n            e.printStackTrace();\r\n        }\r\n    }\r\n};\r\n\r\nclass HelloWorldConsumer : public ExceptionListener,\r\n                           public MessageListener,\r\n                           public Runnable {\r\n\r\nprivate:\r\n\r\n    CountDownLatch latch;\r\n    CountDownLatch doneLatch;\r\n    Connection* connection;\r\n    Session* session;\r\n    Destination* destination;\r\n    MessageConsumer* consumer;\r\n    MessageProducer *producer;\r\n    long waitMillis;\r\n    bool useTopic;\r\n    bool sessionTransacted;\r\n    std::string brokerURI;\r\n\r\nprivate:\r\n\r\n    HelloWorldConsumer(const HelloWorldConsumer&);\r\n    HelloWorldConsumer& operator=(const HelloWorldConsumer&);\r\n\r\npublic:\r\n\r\n    HelloWorldConsumer(const std::string& brokerURI, int numMessages, bool useTopic = false, bool sessionTransacted = false, int waitMillis = 30000) :\r\n        latch(1),\r\n        doneLatch(numMessages),\r\n        connection(NULL),\r\n        session(NULL),\r\n        destination(NULL),\r\n        consumer(NULL),\r\n        producer(NULL),\r\n        waitMillis(waitMillis),\r\n        useTopic(useTopic),\r\n        sessionTransacted(sessionTransacted),\r\n        brokerURI(brokerURI) {\r\n    }\r\n\r\n    virtual ~HelloWorldConsumer() {\r\n        cleanup();\r\n    }\r\n\r\n    void close() {\r\n        this->cleanup();\r\n    }\r\n\r\n    void waitUntilReady() {\r\n        latch.await();\r\n    }\r\n\r\n    virtual void run() {\r\n\r\n        try {\r\n\r\n            \/\/ Create a ConnectionFactory\r\n            auto_ptr<ConnectionFactory> connectionFactory(\r\n                ConnectionFactory::createCMSConnectionFactory(brokerURI));\r\n\r\n            \/\/ Create a Connection\r\n            connection = connectionFactory->createConnection();\r\n            connection->start();\r\n            connection->setExceptionListener(this);\r\n\r\n            \/\/ Create a Session\r\n            if (this->sessionTransacted == true) {\r\n                session = connection->createSession(Session::SESSION_TRANSACTED);\r\n            } else {\r\n                session = connection->createSession(Session::AUTO_ACKNOWLEDGE);\r\n            }\r\n\r\n            \/\/ Create the destination (Topic or Queue)\r\n            if (useTopic) {\r\n                destination = session->createTopic(QUEUE_NAME);\r\n            } else {\r\n                destination = session->createQueue(QUEUE_NAME);\r\n            }\r\n\r\n            producer = session->createProducer();\r\n            producer->setDeliveryMode(DeliveryMode::NON_PERSISTENT);\r\n\r\n            \/\/ Create a MessageConsumer from the Session to the Topic or Queue\r\n            consumer = session->createConsumer(destination);\r\n\r\n            consumer->setMessageListener(this);\r\n\r\n            std::cout.flush();\r\n            std::cerr.flush();\r\n\r\n            \/\/ Indicate we are ready for messages.\r\n            latch.countDown();\r\n\r\n            \/\/ Wait while asynchronous messages come in.\r\n            doneLatch.await();\r\n\r\n        } catch (CMSException& e) {\r\n            printf(\"Consumer onException() CMS Exception occurred.  Shutting down client. \\n\" );\r\n            \/\/ Indicate we are ready for messages.\r\n            latch.countDown();\r\n            e.printStackTrace();\r\n        }\r\n    }\r\n\r\n    \/\/ Called from the consumer since this class is a registered MessageListener.\r\n    virtual void onMessage(const Message* message) {\r\n\r\n        static int count = 0;\r\n\r\n        try {\r\n            count++;\r\n            \r\n            \r\n            \/\/ Create the Thread Id String\r\n            string threadIdStr = Long::toString(Thread::currentThread()->getId());\r\n\r\n            static bool bPrintf=true;\r\n            if(bPrintf)\r\n            {\r\n                bPrintf=false;\r\n                printf(\"consumer Message threadid: %s\\n\",  threadIdStr.c_str());\r\n            }\r\n\r\n            string strReply=\"consumer return  xxx,ThreadID=\"+threadIdStr;\r\n            const TextMessage* textMessage = dynamic_cast<const TextMessage*> (message);\r\n            \r\n            if(NULL==textMessage)\r\n            {\r\n                printf(\"NULL==textMessage\", message->getCMSType().c_str());\r\n\r\n\r\n                \/\/const cms::MapMessage* mapMsg = dynamic_cast<const cms::MapMessage*>(message);\r\n                \/\/if(mapMsg)\r\n                \/\/{\r\n                \/\/    \r\n                \/\/    std::vector<std::string> elements = mapMsg->getMapNames();\r\n                \/\/    std::vector<std::string>::iterator iter = elements.begin();\r\n                \/\/    for(; iter != elements.end() ; ++iter) \r\n                \/\/    {\r\n                \/\/        std::string key = *iter;\r\n                \/\/        cms::Message::ValueType elementType = mapMsg->getValueType(key);\r\n                \/\/        string strxxx;\r\n                \/\/        int cc=0;\r\n                \/\/        switch(elementType) {\r\n                \/\/    case cms::Message::BOOLEAN_TYPE:\r\n                \/\/        \/\/msg->setBoolean(key, mapMsg->getBoolean(key));\r\n                \/\/        break;\r\n                \/\/    case cms::Message::BYTE_TYPE:\r\n                \/\/        \/\/msg->setByte(key, mapMsg->getByte(key));\r\n                \/\/        break;\r\n                \/\/    case cms::Message::BYTE_ARRAY_TYPE:\r\n                \/\/        \/\/msg->setBytes(key, mapMsg->getBytes(key));\r\n                \/\/        break;\r\n                \/\/    case cms::Message::CHAR_TYPE:\r\n                \/\/        \/\/msg->setChar(key, mapMsg->getChar(key));\r\n                \/\/        break;\r\n                \/\/    case cms::Message::SHORT_TYPE:\r\n                \/\/        \/\/msg->setShort(key, mapMsg->getShort(key));\r\n                \/\/        break;\r\n                \/\/    case cms::Message::INTEGER_TYPE:\r\n                \/\/        \/\/msg->setInt(key, mapMsg->getInt(key));\r\n                \/\/        break;\r\n                \/\/    case cms::Message::LONG_TYPE:\r\n                \/\/        \/\/msg->setLong(key, mapMsg->getLong(key));\r\n                \/\/        break;\r\n                \/\/    case cms::Message::FLOAT_TYPE:\r\n                \/\/        \/\/msg->setFloat(key, mapMsg->getFloat(key));\r\n                \/\/        break;\r\n                \/\/    case cms::Message::DOUBLE_TYPE:\r\n                \/\/        \/\/msg->setDouble(key, mapMsg->getDouble(key));\r\n                \/\/        break;\r\n                \/\/    case cms::Message::STRING_TYPE:\r\n                \/\/        \/\/msg->setString(key, mapMsg->getString(key));\r\n                \/\/        strxxx=mapMsg->getString(key);\r\n                \/\/        cc=1;\r\n                \/\/        break;\r\n                \/\/    default:\r\n                \/\/        break;\r\n                \/\/        }\r\n                \/\/    }\r\n\r\n                \/\/}\r\n\r\n                return;\r\n            }\r\n\r\n            std::auto_ptr<TextMessage> responsemessage(session->createTextMessage(strReply));\r\n            responsemessage->setCMSCorrelationID(textMessage->getCMSCorrelationID());\r\n            \r\n            \r\n            string text = \"\";\r\n\r\n            if (textMessage != NULL) {\r\n                text = textMessage->getText();\r\n            } else {\r\n                text = \"NOT A TEXTMESSAGE!\";\r\n            }\r\n            \r\n            int nProPerty=textMessage->getIntProperty(\"Integer\");\r\n            printf(\"consumer Message #%d Received: %s,nProPerty[%d]\\n\", count, text.c_str(),nProPerty);\r\n            \r\n            \r\n            const cms::Destination* destSend=textMessage->getCMSReplyTo();\r\n            if(destSend)\r\n            {\r\n                this->producer->send(destSend,responsemessage.get());\r\n\r\n                printf(\"consumer Message #%d send: %s\\n\", count, strReply.c_str());\r\n            }\r\n            \r\n\r\n        } catch (CMSException& e) {\r\n            printf(\"Consumer onMessage() CMS Exception occurred.  Shutting down client. \\n\" );\r\n            e.printStackTrace();\r\n        }\r\n\r\n        \/\/ Commit all messages.\r\n        if (this->sessionTransacted) {\r\n            session->commit();\r\n        }\r\n\r\n        \/\/ No matter what, tag the count down latch until done.\r\n        \/\/doneLatch.countDown();\r\n    }\r\n\r\n    \/\/ If something bad happens you see it here as this class is also been\r\n    \/\/ registered as an ExceptionListener with the connection.\r\n    virtual void onException(const CMSException& ex AMQCPP_UNUSED) {\r\n        printf(\"Consumer onException() CMS Exception occurred.  Shutting down client. \\n\" );\r\n        \/\/printf(\"CMS Exception occurred.  Shutting down client.\\n\");\r\n        ex.printStackTrace();\r\n        exit(1);\r\n    }\r\n\r\nprivate:\r\n\r\n    void cleanup() {\r\n        if (connection != NULL) {\r\n            try {\r\n                connection->close();\r\n            } catch (cms::CMSException& ex) {\r\n                ex.printStackTrace();\r\n            }\r\n        }\r\n\r\n        \/\/ Destroy resources.\r\n        try {\r\n            delete destination;\r\n            destination = NULL;\r\n            delete consumer;\r\n            consumer = NULL;\r\n            delete session;\r\n            session = NULL;\r\n            delete connection;\r\n            connection = NULL;\r\n        } catch (CMSException& e) {\r\n            e.printStackTrace();\r\n        }\r\n    }\r\n};\r\n\r\nint main(int argc AMQCPP_UNUSED, char* argv[] AMQCPP_UNUSED) {\r\n\r\n    \/\/if(argc<2)\r\n    \/\/{\r\n    \/\/    printf(\"argc<2\\r\\n\");\r\n    \/\/    return 0;\r\n    \/\/}\r\n\r\n    activemq::library::ActiveMQCPP::initializeLibrary();\r\n    {\r\n    std::cout << \"=====================================================\\n\";\r\n    std::cout << \"Starting the example:\" << std::endl;\r\n    std::cout << \"-----------------------------------------------------\\n\";\r\n\r\n\r\n    \/\/ Set the URI to point to the IP Address of your broker.\r\n    \/\/ add any optional params to the url to enable things like\r\n    \/\/ tightMarshalling or tcp logging etc.  See the CMS web site for\r\n    \/\/ a full list of configuration options.\r\n    \/\/\r\n    \/\/  http:\/\/activemq.apache.org\/cms\/\r\n    \/\/\r\n    \/\/ Wire Format Options:\r\n    \/\/ =========================\r\n    \/\/ Use either stomp or openwire, the default ports are different for each\r\n    \/\/\r\n    \/\/ Examples:\r\n    \/\/    tcp:\/\/127.0.0.1:61616                      default to openwire\r\n    \/\/    tcp:\/\/127.0.0.1:61616?wireFormat=openwire  same as above\r\n    \/\/    tcp:\/\/127.0.0.1:61613?wireFormat=stomp     use stomp instead\r\n    \/\/\r\n    \/\/ SSL:\r\n    \/\/ =========================\r\n    \/\/ To use SSL you need to specify the location of the trusted Root CA or the\r\n    \/\/ certificate for the broker you want to connect to.  Using the Root CA allows\r\n    \/\/ you to use failover with multiple servers all using certificates signed by\r\n    \/\/ the trusted root.  If using client authentication you also need to specify\r\n    \/\/ the location of the client Certificate.\r\n    \/\/\r\n    \/\/     System::setProperty( \"decaf.net.ssl.keyStore\", \"<path>\/client.pem\" );\r\n    \/\/     System::setProperty( \"decaf.net.ssl.keyStorePassword\", \"password\" );\r\n    \/\/     System::setProperty( \"decaf.net.ssl.trustStore\", \"<path>\/rootCA.pem\" );\r\n    \/\/\r\n    \/\/ The you just specify the ssl transport in the URI, for example:\r\n    \/\/\r\n    \/\/     ssl:\/\/localhost:61617\r\n    \/\/\r\n    std::string brokerURI =\r\n        \"failover:(tcp:\/\/192.168.10.143:61616\"\r\n\/\/        \"?wireFormat=openwire\"\r\n\/\/        \"&transport.useInactivityMonitor=false\"\r\n\/\/        \"&connection.alwaysSyncSend=true\"\r\n\/\/        \"&connection.useAsyncSend=true\"\r\n\/\/        \"?transport.commandTracingEnabled=true\"\r\n\/\/        \"&transport.tcpTracingEnabled=true\"\r\n\/\/        \"&wireFormat.tightEncodingEnabled=true\"\r\n        \")\";\r\n\r\n    \/\/============================================================\r\n    \/\/ set to true to use topics instead of queues\r\n    \/\/ Note in the code above that this causes createTopic or\r\n    \/\/ createQueue to be used in both consumer an producer.\r\n    \/\/============================================================\r\n    bool useTopics = false;\r\n    bool sessionTransacted = true;\r\n    int numMessages = 1;\r\n    bool useConsumer=true;\r\n    bool useProducer=true;\r\n\r\n    \/\/int nSet=atoi(argv[1]);\r\n    \/\/if(1==nSet)\r\n    \/\/{\r\n        \/\/#define USE_COMSUMER\r\n\r\n        \r\n    \/\/}\r\n    \/\/else\r\n    \/\/{\r\n        \/\/#define USE_PRODUCER\r\n\r\n        \/\/\r\n    \/\/}\r\n\r\n\r\n\r\n    long long startTime = System::currentTimeMillis();\r\n    \r\n#ifdef USE_PRODUCER\r\n    printf(\"\u5f53\u524d USE_PRODUCER \\r\\n\");\r\n\r\n    int numProducerMessages = 30;\r\n    int nThreadNumber=10;\r\n    vector<HelloWorldProducer *> vHelloWorldProducer;\r\n    for(int i=0;i<nThreadNumber;++i)\r\n    {\r\n        HelloWorldProducer * producerTemp=new HelloWorldProducer(brokerURI, numProducerMessages, useTopics);\r\n        vHelloWorldProducer.push_back(producerTemp);\r\n    }\r\n\r\n#endif\r\n    \r\n#ifdef USE_COMSUMER\r\n    printf(\"\u5f53\u524d USE_COMSUMER \\r\\n\");\r\n    HelloWorldConsumer consumer(brokerURI, numMessages, useTopics, sessionTransacted);\r\n    \/\/ Start the consumer thread.\r\n    Thread consumerThread(&consumer);\r\n    consumerThread.start();\r\n\r\n    \/\/ Wait for the consumer to indicate that its ready to go.\r\n    consumer.waitUntilReady();\r\n\r\n#endif\r\n\r\n    \r\n\r\n\r\n#ifdef USE_PRODUCER\r\n    \/\/ Start the producer thread.\r\n\r\n    vector<Thread *> vThread;\r\n    for(int i=0;i<nThreadNumber;++i)\r\n    {\r\n        HelloWorldProducer & ProducerTemp=*vHelloWorldProducer[i];\r\n        Thread * threadTemp=new Thread(&ProducerTemp);\r\n        vThread.push_back(threadTemp);\r\n        threadTemp->start();\r\n        ProducerTemp.waitUntilReady();\r\n\r\n    }\r\n\r\n    for(int i=0;i<vThread.size();++i)\r\n    {\r\n        Thread * threadTemp=vThread[i];\r\n        \/\/threadTemp->join();\r\n    }\r\n    while(1)\r\n    {\r\n        Thread::sleep(10);\r\n    }\r\n\r\n    \/\/Thread producerThread1(&producer1);\r\n    \/\/producerThread1.start();\r\n    \/\/producer1.waitUntilReady();\r\n\r\n    \/\/Thread producerThread2(&producer2);\r\n    \/\/producerThread2.start();\r\n    \/\/producer2.waitUntilReady();\r\n\r\n    \/\/Thread producerThread3(&producer3);\r\n    \/\/producerThread3.start();\r\n    \/\/producer3.waitUntilReady();\r\n#endif\r\n\r\n\r\n\r\n    \r\n#ifdef USE_PRODUCER\r\n    \/\/ Wait for the threads to complete.\r\n    \/\/producerThread1.join();\r\n    \/\/producerThread2.join();\r\n    \/\/producerThread3.join();\r\n#endif\r\n\r\n#ifdef USE_COMSUMER\r\n    consumerThread.join();\r\n#endif\r\n\r\n    long long endTime = System::currentTimeMillis();\r\n    double totalTime = (double)(endTime - startTime) \/ 1000.0;\r\n\r\n#ifdef USE_PRODUCER\r\n    \/\/producer1.close();\r\n    \/\/producer2.close();\r\n    \/\/producer3.close();\r\n\r\n    for(int i=0;i<vHelloWorldProducer.size();++i)\r\n    {\r\n        HelloWorldProducer * ProducerTemp=vHelloWorldProducer[i];\r\n        ProducerTemp->close();\r\n\r\n        if(ProducerTemp)\r\n        {\r\n            delete ProducerTemp;\r\n            ProducerTemp=NULL;\r\n        }\r\n    }\r\n\r\n#endif\r\n#ifdef USE_COMSUMER\r\n    consumer.close();\r\n#endif\r\n\r\n    \r\n   \r\n\r\n    std::cout << \"Time to completion = \" << totalTime << \" seconds.\" << std::endl;\r\n    std::cout << \"-----------------------------------------------------\\n\";\r\n    std::cout << \"Finished with the example.\" << std::endl;\r\n    std::cout << \"=====================================================\\n\";\r\n\r\n    }\r\n    activemq::library::ActiveMQCPP::shutdownLibrary();\r\n\r\n\r\n    return 0;\r\n}\r\n\r\n\/\/ END SNIPPET: demo<\/pre>\n<\/span><\/div>\n<\/div>\n <\/p>\n
\u7a0b\u5e8f\u8fd0\u884c\u7ed3\u679c:<\/p>\n
<\/p>\n
 <\/p>\n
\u5173\u4e8eactivemq-cpp \u7684Message \u6d88\u606f\u8f6c\u6362.<\/strong><\/p>\n
activemq-cpp \u4e2d\u7684\u8f6c\u6362ActiveMQMessageTransformation.transformMessage \u4e2d\u662f\u6709\u76f8\u5e94\u7684\u5b9e\u73b0.<\/p>\n
\n<\/span><\/div>\n\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\r\nbool ActiveMQMessageTransformation::transformMessage(cms::Message* message, ActiveMQConnection* connection, Message** amqMessage) {\r\n\r\n    if (message == NULL) {\r\n        throw NullPointerException(__FILE__, __LINE__, \"Provided source cms::Message pointer was NULL\");\r\n    }\r\n\r\n    if (amqMessage == NULL) {\r\n        throw NullPointerException(__FILE__, __LINE__, \"Provided target commands::Message pointer was NULL\");\r\n    }\r\n\r\n    *amqMessage = dynamic_cast<Message*>(message);\r\n\r\n    if (*amqMessage != NULL) {\r\n        return false;\r\n    } else {\r\n\r\n        if (dynamic_cast<cms::BytesMessage*>(message) != NULL) {\r\n            cms::BytesMessage* bytesMsg = dynamic_cast<cms::BytesMessage*>(message);\r\n            bytesMsg->reset();\r\n            ActiveMQBytesMessage* msg = new ActiveMQBytesMessage();\r\n            msg->setConnection(connection);\r\n            try {\r\n                for (;;) {\r\n                    \/\/ Reads a byte from the message stream until the stream is empty\r\n                    msg->writeByte(bytesMsg->readByte());\r\n                }\r\n            } catch (cms::MessageEOFException& e) {\r\n                \/\/ if an end of message stream as expected\r\n            } catch (cms::CMSException& e) {\r\n            }\r\n\r\n            *amqMessage = msg;\r\n        } else if (dynamic_cast<cms::MapMessage*>(message) != NULL) {\r\n            cms::MapMessage* mapMsg = dynamic_cast<cms::MapMessage*>(message);\r\n            ActiveMQMapMessage* msg = new ActiveMQMapMessage();\r\n            msg->setConnection(connection);\r\n\r\n            std::vector<std::string> elements = mapMsg->getMapNames();\r\n            std::vector<std::string>::iterator iter = elements.begin();\r\n            for(; iter != elements.end() ; ++iter) {\r\n                std::string key = *iter;\r\n                cms::Message::ValueType elementType = mapMsg->getValueType(key);\r\n\r\n                switch(elementType) {\r\n                    case cms::Message::BOOLEAN_TYPE:\r\n                        msg->setBoolean(key, mapMsg->getBoolean(key));\r\n                        break;\r\n                    case cms::Message::BYTE_TYPE:\r\n                        msg->setByte(key, mapMsg->getByte(key));\r\n                        break;\r\n                    case cms::Message::BYTE_ARRAY_TYPE:\r\n                        msg->setBytes(key, mapMsg->getBytes(key));\r\n                        break;\r\n                    case cms::Message::CHAR_TYPE:\r\n                        msg->setChar(key, mapMsg->getChar(key));\r\n                        break;\r\n                    case cms::Message::SHORT_TYPE:\r\n                        msg->setShort(key, mapMsg->getShort(key));\r\n                        break;\r\n                    case cms::Message::INTEGER_TYPE:\r\n                        msg->setInt(key, mapMsg->getInt(key));\r\n                        break;\r\n                    case cms::Message::LONG_TYPE:\r\n                        msg->setLong(key, mapMsg->getLong(key));\r\n                        break;\r\n                    case cms::Message::FLOAT_TYPE:\r\n                        msg->setFloat(key, mapMsg->getFloat(key));\r\n                        break;\r\n                    case cms::Message::DOUBLE_TYPE:\r\n                        msg->setDouble(key, mapMsg->getDouble(key));\r\n                        break;\r\n                    case cms::Message::STRING_TYPE:\r\n                        msg->setString(key, mapMsg->getString(key));\r\n                        break;\r\n                    default:\r\n                        break;\r\n                }\r\n            }\r\n\r\n            *amqMessage = msg;\r\n        } else if (dynamic_cast<cms::ObjectMessage*>(message) != NULL) {\r\n            cms::ObjectMessage* objMsg = dynamic_cast<cms::ObjectMessage*>(message);\r\n            ActiveMQObjectMessage* msg = new ActiveMQObjectMessage();\r\n            msg->setConnection(connection);\r\n            msg->setObjectBytes(objMsg->getObjectBytes());\r\n            *amqMessage = msg;\r\n        } else if (dynamic_cast<cms::StreamMessage*>(message) != NULL) {\r\n            cms::StreamMessage* streamMessage = dynamic_cast<cms::StreamMessage*>(message);\r\n            streamMessage->reset();\r\n            ActiveMQStreamMessage* msg = new ActiveMQStreamMessage();\r\n            msg->setConnection(connection);\r\n\r\n            try {\r\n                while(true) {\r\n                    cms::Message::ValueType elementType = streamMessage->getNextValueType();\r\n                    int result = -1;\r\n                    std::vector<unsigned char> buffer(255);\r\n\r\n                    switch(elementType) {\r\n                        case cms::Message::BOOLEAN_TYPE:\r\n                            msg->writeBoolean(streamMessage->readBoolean());\r\n                            break;\r\n                        case cms::Message::BYTE_TYPE:\r\n                            msg->writeBoolean(streamMessage->readBoolean());\r\n                            break;\r\n                        case cms::Message::BYTE_ARRAY_TYPE:\r\n                            while ((result = streamMessage->readBytes(buffer)) != -1) {\r\n                                msg->writeBytes(&buffer[0], 0, result);\r\n                                buffer.clear();\r\n                            }\r\n                            break;\r\n                        case cms::Message::CHAR_TYPE:\r\n                            msg->writeChar(streamMessage->readChar());\r\n                            break;\r\n                        case cms::Message::SHORT_TYPE:\r\n                            msg->writeShort(streamMessage->readShort());\r\n                            break;\r\n                        case cms::Message::INTEGER_TYPE:\r\n                            msg->writeInt(streamMessage->readInt());\r\n                            break;\r\n                        case cms::Message::LONG_TYPE:\r\n                            msg->writeLong(streamMessage->readLong());\r\n                            break;\r\n                        case cms::Message::FLOAT_TYPE:\r\n                            msg->writeFloat(streamMessage->readFloat());\r\n                            break;\r\n                        case cms::Message::DOUBLE_TYPE:\r\n                            msg->writeDouble(streamMessage->readDouble());\r\n                            break;\r\n                        case cms::Message::STRING_TYPE:\r\n                            msg->writeString(streamMessage->readString());\r\n                            break;\r\n                        default:\r\n                            break;\r\n                    }\r\n                }\r\n            } catch (cms::MessageEOFException& e) {\r\n                \/\/ if an end of message stream as expected\r\n            } catch (cms::CMSException& e) {\r\n            }\r\n\r\n            *amqMessage = msg;\r\n        } else if (dynamic_cast<cms::TextMessage*>(message) != NULL) {\r\n            cms::TextMessage* textMsg = dynamic_cast<cms::TextMessage*>(message);\r\n            ActiveMQTextMessage* msg = new ActiveMQTextMessage();\r\n            msg->setConnection(connection);\r\n            msg->setText(textMsg->getText());\r\n            *amqMessage = msg;\r\n        } else {\r\n            *amqMessage = new ActiveMQMessage();\r\n            (*amqMessage)->setConnection(connection);\r\n        }\r\n\r\n        ActiveMQMessageTransformation::copyProperties(message, dynamic_cast<cms::Message*>(*amqMessage));\r\n    }\r\n\r\n    return true;\r\n}<\/pre>\n<\/span><\/div>\n<\/div>\n <\/p>\n
 <\/p>\n
 <\/p>\n
5,activemq \u7684activemq broker cluster (activemq \u96c6\u7fa4).<\/h1>\n <\/p>\n
\u53ef\u4ee5\u53c2\u8003:<\/p>\n
http:\/\/bh-keven.iteye.com\/blog\/1617788<\/p>\n
http:\/\/blog.csdn.net\/jason5186\/article\/details\/18702523<\/p>\n
 <\/p>\n
6,activemq.xml \u4e2d\u7684\u914d\u7f6e\u548cactivemq Connection URIS \u914d\u7f6e<\/h1>\n <\/p>\n
Index\u00a0>\u00a0Apache.NMS.ActiveMQ\u00a0>\u00a0ActiveMQ URI Configuration<\/div>\nhttp:\/\/activemq.apache.org\/nms\/activemq-uri-configuration.html<\/p>\n
http:\/\/activemq.apache.org\/tcp-transport-reference.html<\/p>\n
 <\/p>\n
\u662f\u6709\u76f8\u5e94\u4ecb\u7ecd,\u4f46\u9700\u8981\u82b1\u4e00\u4e9b\u65f6\u95f4\u53bb\u8bfb.<\/p>\n
 <\/p>\n
 <\/p>\n
 <\/p>\n
\/\/7,wireFormat=openwire \u7684\u51e0\u79cd\u65b9\u5f0f.\u7684\u4f18\u7f3a\u70b9.<\/p>\n
\/\/openwire,amqp,stomp,mqtt,ws<\/p>\n","protected":false},"excerpt":{"rendered":"
\u5f88\u65e9\u524d\u5c31\u4ed3\u4fc3\u7684\u63a5\u89e6\u8fc7activemq,\u4f46\u5f53\u65f6\u592a\u8d76\u65f6\u95f4.\u540e\u9762\u53d1\u73b0activemq  …<\/p>\n
\u7ee7\u7eed\u9605\u8bfb »<\/a><\/p>\n","protected":false},"author":1,"featured_media":1015,"comment_status":"closed","ping_status":"closed","sticky":false,"template":"","format":"standard","meta":{"footnotes":""},"categories":[],"tags":[63],"class_list":["post-1014","post","type-post","status-publish","format-standard","has-post-thumbnail","hentry","tag-c"],"views":444,"_links":{"self":[{"href":"http:\/\/zerobox.org\/notes\/wp-json\/wp\/v2\/posts\/1014","targetHints":{"allow":["GET"]}}],"collection":[{"href":"http:\/\/zerobox.org\/notes\/wp-json\/wp\/v2\/posts"}],"about":[{"href":"http:\/\/zerobox.org\/notes\/wp-json\/wp\/v2\/types\/post"}],"author":[{"embeddable":true,"href":"http:\/\/zerobox.org\/notes\/wp-json\/wp\/v2\/users\/1"}],"replies":[{"embeddable":true,"href":"http:\/\/zerobox.org\/notes\/wp-json\/wp\/v2\/comments?post=1014"}],"version-history":[{"count":0,"href":"http:\/\/zerobox.org\/notes\/wp-json\/wp\/v2\/posts\/1014\/revisions"}],"wp:featuredmedia":[{"embeddable":true,"href":"http:\/\/zerobox.org\/notes\/wp-json\/wp\/v2\/media\/1015"}],"wp:attachment":[{"href":"http:\/\/zerobox.org\/notes\/wp-json\/wp\/v2\/media?parent=1014"}],"wp:term":[{"taxonomy":"category","embeddable":true,"href":"http:\/\/zerobox.org\/notes\/wp-json\/wp\/v2\/categories?post=1014"},{"taxonomy":"post_tag","embeddable":true,"href":"http:\/\/zerobox.org\/notes\/wp-json\/wp\/v2\/tags?post=1014"}],"curies":[{"name":"wp","href":"https:\/\/api.w.org\/{rel}","templated":true}]}}
3,activemq\u7684\u51e0\u79cd\u901a\u4fe1\u6a21\u5f0f.<\/h1>\n\u53ef\u4ee5\u53c2\u8003:<\/p>\nhttp:\/\/shmilyaw-hotmail-com.iteye.com\/blog\/1897635<\/p>\n <\/p>\n\u76ee\u524d\u672c\u4eba\u9700\u8981\u7684\u662factivemq-cpp\u7684\u00a0request-response \u6a21\u5f0f.<\/p>\n <\/p>\n <\/p>\n <\/p>\n

5,activemq \u7684activemq broker cluster (activemq \u96c6\u7fa4).<\/h1>\n <\/p>\n\u53ef\u4ee5\u53c2\u8003:<\/p>\nhttp:\/\/bh-keven.iteye.com\/blog\/1617788<\/p>\nhttp:\/\/blog.csdn.net\/jason5186\/article\/details\/18702523<\/p>\n <\/p>\n

3,activemq\u7684\u51e0\u79cd\u901a\u4fe1\u6a21\u5f0f.<\/h1>\n
\u53ef\u4ee5\u53c2\u8003:<\/p>\n
http:\/\/shmilyaw-hotmail-com.iteye.com\/blog\/1897635<\/p>\n
<\/p>\n
\u76ee\u524d\u672c\u4eba\u9700\u8981\u7684\u662factivemq-cpp\u7684\u00a0request-response \u6a21\u5f0f.<\/p>\n
<\/p>\n
<\/p>\n
<\/p>\n

5,activemq \u7684activemq broker cluster (activemq \u96c6\u7fa4).<\/h1>\n
<\/p>\n
\u53ef\u4ee5\u53c2\u8003:<\/p>\n
http:\/\/bh-keven.iteye.com\/blog\/1617788<\/p>\n
http:\/\/blog.csdn.net\/jason5186\/article\/details\/18702523<\/p>\n
<\/p>\n
