使用ActiveMQ进行C++与C#的通信4 - 使用C++连接ActiveMQ
2017-08-07 17:43
375 查看
在上一节编译ActiveMQ-CPP的基础上,创建C++控制台应用程序,将activemq-cpp项目中的include文件夹拷贝到该C++项目中,设置好附加包含目录。
将生成好的libactivemq-cppd.dll、libactivemq-cppd.lib拷贝到当前项目,设置好附加库目录和附加依赖项。
参考activemq-cpp\src\examples\producers\SimpleProducer.cpp中的生产者模型代码,得到如下发送消息代码:
将生成好的libactivemq-cppd.dll、libactivemq-cppd.lib拷贝到当前项目,设置好附加库目录和附加依赖项。
参考activemq-cpp\src\examples\producers\SimpleProducer.cpp中的生产者模型代码,得到如下发送消息代码:
/* * Licensed to the Apache Software Foundation (ASF) under one or more * contributor license agreements. See the NOTICE file distributed with * this work for additional information regarding copyright ownership. * The ASF licenses this file to You under the Apache License, Version 2.0 * (the "License"); you may not use this file except in compliance with * the License. You may obtain a copy of the License at * * http://www.apache.org/licenses/LICENSE-2.0 * * Unless required by applicable law or agreed to in writing, software * distributed under the License is distributed on an "AS IS" BASIS, * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. * See the License for the specific language governing permissions and * limitations under the License. */ // START SNIPPET: demo #include "stdafx.h" #include <activemq/library/ActiveMQCPP.h> #include <decaf/lang/Thread.h> #include <decaf/lang/Runnable.h> #include <decaf/util/concurrent/CountDownLatch.h> #include <decaf/lang/Integer.h> #include <decaf/lang/Long.h> #include <decaf/lang/System.h> #include <activemq/core/ActiveMQConnectionFactory.h> #include <activemq/util/Config.h> #include <cms/Connection.h> #include <cms/Session.h> #include <cms/TextMessage.h> #include <cms/BytesMessage.h> #include <cms/MapMessage.h> #include <cms/ExceptionListener.h> #include <cms/MessageListener.h> #include <stdlib.h> #include <stdio.h> #include <iostream> #include <memory> using namespace activemq::core; using namespace decaf::util::concurrent; using namespace decaf::util; using namespace decaf::lang; using namespace cms; using namespace std; #pragma comment(lib, "..\\ActiveMQ-cpp\\lib\\Debug\\libactivemq-cppd.lib") class HelloWorldProducer : public Runnable { private: Connection* connection; Session* session; Destination* destination; MessageProducer* producer; int numMessages; bool useTopic; bool sessionTransacted; std::string brokerURI; private: HelloWorldProducer(const HelloWorldProducer&); HelloWorldProducer& operator=(const HelloWorldProducer&); public: HelloWorldProducer(const std::string& brokerURI, int numMessages, bool useTopic = false, bool sessionTransacted = false) : connection(NULL), session(NULL), destination(NULL), producer(NULL), numMessages(numMessages), useTopic(useTopic), sessionTransacted(sessionTransacted), brokerURI(brokerURI) { } virtual ~HelloWorldProducer(){ cleanup(); } void close() { this->cleanup(); } virtual void run() { try { // Create a ConnectionFactory brokerURI = "tcp://localhost:61616?wireFormat=openwire&connection.useAsyncSend=true"; auto_ptr<ConnectionFactory> connectionFactory( ConnectionFactory::createCMSConnectionFactory(brokerURI)); // Create a Connection connection = connectionFactory->createConnection("system", "manager"); connection->start(); // Create a Session // if (this->sessionTransacted) { // session = connection->createSession(Session::SESSION_TRANSACTED); // } else { // session = connection->createSession(Session::AUTO_ACKNOWLEDGE); // } session = connection->createSession(); // Create the destination (Topic or Queue) if (useTopic) { destination = session->createTopic("test.log"); } else { destination = session->createQueue("test.log"); } // Create a MessageProducer from the Session to the Topic or Queue producer = session->createProducer(destination); producer->setDeliveryMode(DeliveryMode::NON_PERSISTENT); // Create the Thread Id String string threadIdStr = Long::toString(Thread::currentThread()->getId()); // Create a messages string text = (string) "Hello world! from thread " + threadIdStr; for (int ix = 0; ix < numMessages; ++ix) { std::auto_ptr<TextMessage> message(session->createTextMessage(text)); message->setIntProperty("Integer", ix); printf("Sent message #%d from thread %s\n", ix + 1, threadIdStr.c_str()); producer->send(message.get()); } } catch (CMSException& e) { e.printStackTrace(); } } private: void cleanup() { if (connection != NULL) { try { connection->close(); } catch (cms::CMSException& ex) { ex.printStackTrace(); } } // Destroy resources. try { delete destination; destination = NULL; delete producer; producer = NULL; delete session; session = NULL; delete connection; connection = NULL; } catch (CMSException& e) { e.printStackTrace(); } } }; class HelloWorldConsumer : public ExceptionListener, public MessageListener, public Runnable { private: CountDownLatch latch; CountDownLatch doneLatch; Connection* connection; Session* session; Destination* destination; MessageConsumer* consumer; long waitMillis; bool useTopic; bool sessionTransacted; std::string brokerURI; private: HelloWorldConsumer(const HelloWorldConsumer&); HelloWorldConsumer& operator=(const HelloWorldConsumer&); public: HelloWorldConsumer(const std::string& brokerURI, int numMessages, bool useTopic = false, bool sessionTransacted = false, int waitMillis = 30000) : latch(1), doneLatch(numMessages), connection(NULL), session(NULL), destination(NULL), consumer(NULL), waitMillis(waitMillis), useTopic(useTopic), sessionTransacted(sessionTransacted), brokerURI(brokerURI) { } virtual ~HelloWorldConsumer() { cleanup(); } void close() { this->cleanup(); } void waitUntilReady() { latch.await(); } virtual void run() { try { // Create a ConnectionFactory auto_ptr<ConnectionFactory> connectionFactory( ConnectionFactory::createCMSConnectionFactory(brokerURI)); // Create a Connection connection = connectionFactory->createConnection("system", "manager"); connection->start(); connection->setExceptionListener(this); // Create a Session if (this->sessionTransacted == true) { session = connection->createSession(Session::SESSION_TRANSACTED); } else { session = connection->createSession(Session::AUTO_ACKNOWLEDGE); } // Create the destination (Topic or Queue) if (useTopic) { destination = session->createTopic("TEST.FOO"); } else { destination = session->createQueue("TEST.FOO"); } // Create a MessageConsumer from the Session to the Topic or Queue consumer = session->createConsumer(destination); consumer->setMessageListener(this); std::cout.flush(); std::cerr.flush(); // Indicate we are ready for messages. latch.countDown(); // Wait while asynchronous messages come in. doneLatch.await(waitMillis); } catch (CMSException& e) { // Indicate we are ready for messages. latch.countDown(); e.printStackTrace(); } } // Called from the consumer since this class is a registered MessageListener. virtual void onMessage(const Message* message) { static int count = 0; try { count++; const TextMessage* textMessage = dynamic_cast<const TextMessage*> (message); string text = ""; if (textMessage != NULL) { text = textMessage->getText(); } else { text = "NOT A TEXTMESSAGE!"; } printf("Message #%d Received: %s\n", count, text.c_str()); } catch (CMSException& e) { e.printStackTrace(); } // Commit all messages. if (this->sessionTransacted) { session->commit(); } // No matter what, tag the count down latch until done. doneLatch.countDown(); } // If something bad happens you see it here as this class is also been // registered as an ExceptionListener with the connection. virtual void onException(const CMSException& ex AMQCPP_UNUSED) { printf("CMS Exception occurred. Shutting down client.\n"); ex.printStackTrace(); exit(1); } private: void cleanup() { if (connection != NULL) { try { connection->close(); } catch (cms::CMSException& ex) { ex.printStackTrace(); } } // Destroy resources. try { delete destination; destination = NULL; delete consumer; consumer = NULL; delete session; session = NULL; delete connection; connection = NULL; } catch (CMSException& e) { e.printStackTrace(); } } }; int main(int argc AMQCPP_UNUSED, char* argv[] AMQCPP_UNUSED) { activemq::library::ActiveMQCPP::initializeLibrary(); { std::cout << "=====================================================\n"; std::cout << "Starting the example:" << std::endl; std::cout << "-----------------------------------------------------\n"; // Set the URI to point to the IP Address of your broker. // add any optional params to the url to enable things like // tightMarshalling or tcp logging etc. See the CMS web site for // a full list of configuration options. // // http://activemq.apache.org/cms/ // // Wire Format Options: // ========================= // Use either stomp or openwire, the default ports are different for each // // Examples: // tcp://127.0.0.1:61616 default to openwire // tcp://127.0.0.1:61616?wireFormat=openwire same as above // tcp://127.0.0.1:61613?wireFormat=stomp use stomp instead // // SSL: // ========================= // To use SSL you need to specify the location of the trusted Root CA or the // certificate for the broker you want to connect to. Using the Root CA allows // you to use failover with multiple servers all using certificates signed by // the trusted root. If using client authentication you also need to specify // the location of the client Certificate. // // System::setProperty( "decaf.net.ssl.keyStore", "<path>/client.pem" ); // System::setProperty( "decaf.net.ssl.keyStorePassword", "password" ); // System::setProperty( "decaf.net.ssl.trustStore", "<path>/rootCA.pem" ); // // The you just specify the ssl transport in the URI, for example: // // ssl://localhost:61617 // std::string brokerURI = "failover:(tcp://localhost:61616" // "?wireFormat=openwire" // "&transport.useInactivityMonitor=false" // "&connection.alwaysSyncSend=true" // "&connection.useAsyncSend=true" // "?transport.commandTracingEnabled=true" // "&transport.tcpTracingEnabled=true" // "&wireFormat.tightEncodingEnabled=true" ")"; //============================================================ // set to true to use topics instead of queues // Note in the code above that this causes createTopic or // createQueue to be used in both consumer an producer. //============================================================ bool useTopics = false; bool sessionTransacted = false; int numMessages = 2000; long long startTime = System::currentTimeMillis(); HelloWorldProducer producer(brokerURI, numMessages, useTopics); //HelloWorldConsumer consumer(brokerURI, numMessages, useTopics, sessionTransacted); // Start the consumer thread. // Thread consumerThread(&consumer); // consumerThread.start(); // Wait for the consumer to indicate that its ready to go. //consumer.waitUntilReady(); // Start the producer thread. Thread producerThread(&producer); producerThread.start(); // Wait for the threads to complete. producerThread.join(); //consumerThread.join(); long long endTime = System::currentTimeMillis(); double totalTime = (double)(endTime - startTime) / 1000.0; //consumer.close(); producer.close(); std::cout << "Time to completion = " << totalTime << " seconds." << std::endl; std::cout << "-----------------------------------------------------\n"; std::cout << "Finished with the example." << std::endl; std::cout << "=====================================================\n"; } activemq::library::ActiveMQCPP::shutdownLibrary(); //TODO 异步 int n = 0; cin >> n; } // END SNIPPET: demo
相关文章推荐
- 使用ActiveMQ进行C++与C#的通信2 - 使用C#连接ActiveMQ
- 使用ActiveMQ进行C++与C#的通信5 - 实现C++和C#的通信
- 使用ActiveMQ进行C++与C#的通信1 - ActiveMQ入门
- 使用ActiveMQ进行C++与C#的通信6 - 相关资源(终)
- 用ActiveMQ进行C++与C#的通信3 - 编译ActiveMQ-CPP
- C++ VS 2008中ATL的创建,使用。并在C++项目以及C#项目中进行调用测试
- 1ActiveMQ使用 2ActiveMQC#Client实现 3ActiveMQC++Client实现通讯 4 ActiveMQJavaClient实现
- C#中使用SendMessage进行进程通信的实例
- 进程通信之一 使用WM_COPYDATA C++及C#实现
- C++ VS 2008中ATL的创建,使用。并在C++项目以及C#项目中进行调用测试
- C#使用 UdpClient 类进行简单通信的例子
- 在Javascript中使用protobuf与c++进行通信
- C#使用匿名管道在本地进程之间进行通信
- 在Javascript中使用protobuf与c++进行通信
- C++ VS 2008中ATL的创建,使用。并在C++项目以及C#项目中进行调用测试
- C#中使用命名管道进行进程通信的实例
- Socket编程---使用无连接的数据报(UDP) 进行通信(转)
- 使用C#调用PI-SDK进行基于PI的开发——使用PI-SDK建立与PI数据库的连接(转载)
- [转]如何借助 TLS/SSL 确保套接字连接的安全(使用 C#/VB/C++ 和 XAML 的 Windows 应用商店应用)
- 使用命名管道通过网络在进程之间进行通信(C#)