如何在Linux环境中的Qt项目中使用ActiveMQ-CPP
之前在Linux下的qt程序中使用activeMQ的时候也是用了很多时间去研究，本来想的是好好记录一下，但是当时顾着写代码。很多细节也不想再去走一遍了。大概写一下怎么使用就行了。注意：一定要先开启服务器。
代码1：消费者
代码；这个代码其实是官网的源码里面提供的例子
我来对这个代码做个解释，也是作为自己的笔记。
#include <decaf/lang/Thread.h>
#include <decaf/lang/Runnable.h>
#include <decaf/util/concurrent/CountDownLatch.h>
#include <activemq/core/ActiveMQConnectionFactory.h>
#include <activemq/core/ActiveMQConnection.h>
#include <activemq/transport/DefaultTransportListener.h>
#include <activemq/library/ActiveMQCPP.h>
#include <decaf/lang/Integer.h>
#include <activemq/util/Config.h>
#include <decaf/util/Date.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>
using namespace activemq;
using namespace activemq::core;
using namespace activemq::transport;
using namespace decaf::lang;
using namespace decaf::util;
using namespace decaf::util::concurrent;
using namespace cms;
using namespace std;
class SimpleAsyncConsumer : public ExceptionListener,
public MessageListener,
public DefaultTransportListener {
private:
Connection* connection;
Session* session;
Destination* destination;
MessageConsumer* consumer;
bool useTopic;
std::string brokerURI;
std::string destURI;
bool clientAck;
private:
SimpleAsyncConsumer(const SimpleAsyncConsumer&);
SimpleAsyncConsumer& operator=(const SimpleAsyncConsumer&);
public:
SimpleAsyncConsumer(const std::string& brokerURI,
const std::string& destURI,
bool useTopic = false,
bool clientAck = false) :
connection(NULL),
session(NULL),
destination(NULL),
consumer(NULL),
useTopic(useTopic),
brokerURI(brokerURI),
destURI(destURI),
clientAck(clientAck) {
}
virtual ~SimpleAsyncConsumer() {
this->cleanup();
}
void close() {
this->cleanup();
}
void runConsumer() {
try {
// Create a ConnectionFactory
ActiveMQConnectionFactory* connectionFactory = new ActiveMQConnectionFactory(brokerURI);
// Create a Connection
connection = connectionFactory->createConnection();
delete connectionFactory;
ActiveMQConnection* amqConnection = dynamic_cast<ActiveMQConnection*>(connection);
if (amqConnection != NULL) {
amqConnection->addTransportListener(this);
}
//如果不是使用的failover连接的话。如果服务器没有开启，执行这一步的时候会抛出错误
//如果是使用failover连接，执行到这一步的时候，就不会往下继续执行了，就一直卡在这里，直到和服务器建立连接
//failover连接是一个什么模式我也不是很清楚，我只知道在使用这个模式连接的时候，它是会自动重连的。即使服务器中途断了，服务器再 次开启的时候，它是能自动连接上的
connection->start();
//监听异常。有异常抛出的时候会执行相关的函数
connection->setExceptionListener(this);
// Create a Session
if (clientAck) {
session = connection->createSession(Session::CLIENT_ACKNOWLEDGE);
} else {
session = connection->createSession(Session::AUTO_ACKNOWLEDGE);
}
// Create the destination (Topic or Queue)
//使用queue模式还是topic模式
//queue模式是一对一连接：一个生成者一个消费者。
//topic模式是多对多连接；可以有多个生成者和多个消费者
//如果你需要既发送有接收的消息的话，这个情况大概率是应该使用topic模式的
//在使用topic模式的时候，你自己通过生成者发出去的消息，你自己的消费者也是会收到这个消息的
//这个时候，你可以设置消息过滤，或者使用消息属性识别。
if (useTopic) {
destination = session->createTopic(destURI);
} else {
destination = session->createQueue(destURI);
}
// Create a MessageConsumer from the Session to the Topic or Queue
consumer = session->createConsumer(destination);
consumer->setMessageListener(this);
} catch (CMSException& e) {
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!";
}
if (clientAck) {
message->acknowledge();
}
printf("Message #%d Received: %s\n", count, text.c_str());
} catch (CMSException& e) {
e.printStackTrace();
}
}
// 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");
exit(1);
}
virtual void onException(const decaf::lang::Exception& ex) {
printf("Transport Exception occurred: %s \n", ex.getMessage().c_str());
}
virtual void transportInterrupted() {
std::cout << "The Connection's Transport has been Interrupted." << std::endl;
}
virtual void transportResumed() {
std::cout << "The Connection's Transport has been Restored." << std::endl;
}
private:
void cleanup(){
//*************************************************
// Always close destination, consumers and producers before
// you destroy their sessions and connection.
//*************************************************
// Destroy resources.
try{
if( destination != NULL ) delete destination;
}catch (CMSException& e) {}
destination = NULL;
try{
if( consumer != NULL ) delete consumer;
}catch (CMSException& e) {}
consumer = NULL;
// Close open resources.
try{
if( session != NULL ) session->close();
if( connection != NULL ) connection->close();
}catch (CMSException& e) {}
// Now Destroy them
try{
if( session != NULL ) delete session;
}catch (CMSException& e) {}
session = NULL;
try{
if( connection != NULL ) delete connection;
}catch (CMSException& e) {}
connection = NULL;
}
};
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 IPAddress 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
//
std::string brokerURI =
"failover:(tcp://127.0.0.1:61616"
// "?wireFormat=openwire"
// "&connection.useAsyncSend=true"
// "&transport.commandTracingEnabled=true"
// "&transport.tcpTracingEnabled=true"
// "&wireFormat.tightEncodingEnabled=true"
")";
//============================================================
// This is the Destination Name and URI options. Use this to
// customize where the consumer listens, to have the consumer
// use a topic or queue set the 'useTopics' flag.
//============================================================
std::string destURI = "TEST.FOO"; //?consumer.prefetchSize=1";
//============================================================
// set to true to use topics instead of queues
// Note in the code above that this causes createTopic or
// createQueue to be used in the consumer.
//============================================================
bool useTopics = false;
//============================================================
// set to true if you want the consumer to use client ack mode
// instead of the default auto ack mode.
//============================================================
bool clientAck = false;
// Create the consumer
SimpleAsyncConsumer consumer( brokerURI, destURI, useTopics, clientAck );
// Start it up and it will listen forever.
consumer.runConsumer();
// Wait to exit.
std::cout << "Press 'q' to quit" << std::endl;
while( std::cin.get() != 'q') {}
// All CMS resources should be closed before the library is shutdown.
consumer.close();
std::cout << "-----------------------------------------------------\n";
std::cout << "Finished with the example." << std::endl;
std::cout << "=====================================================\n";
activemq::library::ActiveMQCPP::shutdownLibrary();
}
代码2：生成者
#include <decaf/lang/Thread.h>
#include <decaf/lang/Runnable.h>
#include <decaf/util/concurrent/CountDownLatch.h>
#include <decaf/lang/Long.h>
#include <decaf/util/Date.h>
#include <activemq/core/ActiveMQConnectionFactory.h>
#include <activemq/util/Config.h>
#include <activemq/library/ActiveMQCPP.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;
using namespace activemq::core;
using namespace decaf;
using namespace decaf::lang;
using namespace decaf::util;
using namespace decaf::util::concurrent;
using namespace cms;
using namespace std;
class SimpleProducer : public Runnable {
private:
Connection* connection;
Session* session;
Destination* destination;
MessageProducer* producer;
bool useTopic;
bool clientAck;
unsigned int numMessages;
std::string brokerURI;
std::string destURI;
private:
SimpleProducer( const SimpleProducer& );
SimpleProducer& operator= ( const SimpleProducer& );
public:
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),
useTopic(useTopic),
clientAck(clientAck),
numMessages(numMessages),
brokerURI(brokerURI),
destURI(destURI) {
}
virtual ~SimpleProducer(){
cleanup();
}
void close() {
this->cleanup();
}
virtual void run() {
try {
// Create a ConnectionFactory
auto_ptr<ActiveMQConnectionFactory> connectionFactory(
new ActiveMQConnectionFactory( brokerURI ) );
// Create a Connection
try{
connection = connectionFactory->createConnection();
connection->start();
} catch( CMSException& e ) {
e.printStackTrace();
throw e;
}
// Create a Session
if( clientAck ) {
session = connection->createSession( Session::CLIENT_ACKNOWLEDGE );
} else {
session = connection->createSession( Session::AUTO_ACKNOWLEDGE );
}
// Create the destination (Topic or Queue)
if( useTopic ) {
destination = session->createTopic( destURI );
} else {
destination = session->createQueue( destURI );
}
// 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( unsigned int ix=0; ix<numMessages; ++ix ){
TextMessage* message = session->createTextMessage( text );
//这里有一个setIntProperty，应该也会有一个对应的getIntProperty（我没有去看有没有）。你可以在消费者获取消息的时候调用这个函数，函数属性是否一致，如果一致说明是自己发送的消息，就可以不去处理它。
//可以确认的是TextMessage是有一个setPropertyString和getPropertyString函数的，这个函数可以用了区别是否是自己的消息。
message->setIntProperty( "Integer", ix );
// Tell the producer to send the message
printf( "Sent message #%d from thread %s\n", ix+1, threadIdStr.c_str() );
producer->send( message );
delete message;
}
}catch ( CMSException& e ) {
e.printStackTrace();
}
}
private:
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.printStackTrace(); }
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 = NULL;
try{
if( connection != NULL ) delete connection;
}catch ( CMSException& e ) { e.printStackTrace(); }
connection = NULL;
}
};
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 IPAddress 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
//
std::string brokerURI =
"failover://(tcp://127.0.0.1:61616"
// "?wireFormat=openwire"
// "&connection.useAsyncSend=true"
// "&transport.commandTracingEnabled=true"
// "&transport.tcpTracingEnabled=true"
// "&wireFormat.tightEncodingEnabled=true"
")";
//============================================================
// Total number of messages for this producer to send.
//============================================================
unsigned int numMessages = 2000;
//============================================================
// This is the Destination Name and URI options. Use this to
// customize where the Producer produces, to have the producer
// use a topic or queue set the 'useTopics' flag.
//============================================================
std::string destURI = "TEST.FOO";
//============================================================
// set to true to use topics instead of queues
// Note in the code above that this causes createTopic or
// createQueue to be used in the producer.
//============================================================
bool useTopics = false;
// Create the producer and run it.
SimpleProducer producer( brokerURI, numMessages, destURI, useTopics );
// Publish the given number of Messages
producer.run();
// Before exiting we ensure that all CMS resources are closed.
producer.close();
std::cout << "-----------------------------------------------------\n";
std::cout << "Finished with the example." << std::endl;
std::cout << "=====================================================\n";
activemq::library::ActiveMQCPP::shutdownLibrary();
}
原文地址：https://blog.csdn.net/qq_43611366/article/details/143385887
免责声明：本站文章内容转载自网络资源，如本站内容侵犯了原著者的合法权益，可联系本站删除。更多内容请关注自学内容网（zxcms.com）！