您的位置:首页 > 运维架构 > Linux

Linux 下的 ActiveMQ C++ 客户端开发

2017-11-15 22:59 387 查看
        ActiveMQ CPP是一个提供消息库,即ActiveMQ C++客户端 API。通过该接口,我们可以使用多种协议来与消息中间件进行通信。本文主要针对
CentOS 7  x64(Linux)环境 ActiveMQ CPP的编译及简单客户端开发说明,其他类
Unix系统仅供参考,不要直接照搬本说明!安装软件前,一定要认真阅读软件包中的 README文件,该文件中有详细的前置依赖条件和编译过程说明。

     本文针对ActiveMQ-CPP 3.9.4,其他版本可以参考。

1.安装

       为了在Linux或者其他类Unix系统上编译安装 ActiveMQ-CPP,我们需要安装必要的依赖,下表是 ActiveMQ-CPP
 README 文件列出的依赖,由于操作系统环境的差异,实际上可能不全,需根据需要增加依赖。

Tool

Recommended Version

autoconf

>= 2.61(CentOS系统自带,此处不作说明)

automake

>= 1.10 (CentOS系统自带,此处不作说明)

libtool

>= 1.5.24(CentOS系统自带,此处不作说明)

APR

>= 1.3*

CPPUnit

>= 1.10.2* (推荐1.12.1)

OpenSSL

|>= 0.9.8m* (推荐1.0.0或更高版本,这是一个可选依赖)

   *表示其相应的开发包也需要安装

1.1下载依赖库和activemq-cpp

1)  apr(1.6.3):

https://mirrors.tuna.tsinghua.edu.cn/apache//apr/  (http://apr.apache.org/)

2)   libexpat(2.2.5): 
https://github.com/libexpat/libexpat/tree/R_2_2_5

3)  apr-util(1.6.1):
https://mirrors.tuna.tsinghua.edu.cn/apache//apr/  (http://apr.apache.org/)

4)  apr-iconv(1.2.2): 
https://mirrors.tuna.tsinghua.edu.cn/apache//apr/  (http://apr.apache.org/)

5)   CPPUnit(1.12.1): 
https://sourceforge.net/projects/cppunit/files/cppunit/

6)  OpenSSL:
这个软件是可以选择安装的,如果业务需要用到SSL传输的话,则需要安装。官方推荐使用1.0.0或者更高版本的OpenSSL,具体安装方法请参考OpenSSL的文档;如果不需要,则可选择不安装。

7)  activemq-cpp(3.9.4): 

https://archive.apache.org/dist/activemq/activemq-cpp/(http://activemq.apache.org/cms/download.html)

1.2编译

         由于C++客户端的开发需要依赖以上库,所以,在编译C++客户端程序时,首先需要编译这些依赖库。

1)       apr

下载后解压(tar –xvf ***.tar.gz)进入目录,三部曲:

./configure --prefix=/usr/local/apr/ 

make 

make install

2)   libexpat

下载后解压进入 expat 目录,三部曲:

./buildconf.sh

./configure 

(采用默认路径,libexpat into`/usr/local/lib`, `expat.h` into `/usr/local/include`, and `xmlwf` into`/usr/local/bin`;如果想安装到其他路径,如`/home/me/mystuff/lib`,`/home/me/mystuff/include`, and `/home/me/mystuff/bin`, 可配置为:

./configure--prefix=/home/me/mystuff )

make

3)   apr-util

下载后解压进入目录,三部曲:

./configure --prefix=/usr/local/aprutil--with-apr=/usr/local/apr/ 

make 

make install

4)       apr-iconv

下载后解压进入目录,三部曲:

./configure --prefix=/usr/local/apr-iconv/--with-apr=/usr/local/apr/ 

make 

make install

5)       cppunit

下载后解压进入目录,三部曲:

./configure --prefix=/usr/local/cppunit/

make 

make install(install需要root权限)

执行完后在/usr/local/cppunit/目录下可以看到头文件和库文件

6)       ActiveMQ-CPP

解压编译,由于不需要SSL,因此需加上“
--disable-ssl
”,否则会编译不通过:

./configure --prefix=/usr/local/ActiveMQ-CPP--disable-ssl --with-apr=/usr/local/apr/ --with-apr-util=/usr/local/aprutil/ --with-cppunit=/usr/local/cppunit 

make 

make install

至此编译工作完成,在/usr/local目录下生成了5个目录,分别为ActiveMQ-CPP、apr、apr-iconv、aprutil、cppunit。

2.编程测试

     下面进行简单的测试,示例代码如下 demo.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 <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;

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
auto_ptr<ConnectionFactory> connectionFactory(
ConnectionFactory::createCMSConnectionFactory(brokerURI));

// Create a Connection
connection = connectionFactory->createConnection();
connection->start();

// Create a Session
if (this->sessionTransacted) {
session = connection->createSession(Session::SESSION_TRANSACTED);
} else {
session = connection->createSession(Session::AUTO_ACKNOWLEDGE);
}

// Create the destination (Topic or Queue)
destination = session->createQueue("TEST.FOO");

// 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();
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)

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: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)";

//============================================================
// 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 = true;
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();
}
// END SNIPPET: demo

编写一个Makefile文件,方便以后每次进行编译:

SRCS = $(wildcard *.cpp)
OBJS = $(SRCS:.cpp = .o)
CC = g++
INCLUDES = -I/usr/local/ActiveMQ-CPP/include/activemq-cpp-3.9.4 -I/usr/local/apr/include/apr-1
LIBS = -L/usr/local/ActiveMQ-CPP/lib
LINKS = -lactivemq-cpp
CCFLAGS = -g -Wall -o0
TARGET = demo

$(TARGET):$(OBJS)
$(CC) $^ -o $@ $(INCLUDES) $(LIBS) $(LINKS)
%.o : %.c
$(CC) -c $^ $(CCFLAGS)
clean:
rm $(TARGET)
.PHONY:clean


       Makefile文件中的头文件路径和库文件路径需与编译设置的路径一致,注意以下是笔者的安装相应文件的路径,仅供参考:

        # activemq-cpp-3.9.4、apr的头文件路径和库文件路径

        /usr/local/include/activemq-cpp-3.9.4/

        /usr/apr/include/apr-1

        /usr/local/ActiveMQ-CPP/lib

 

      编译该程序,如果所有都安装正确并且配置正确的话,应该不会得到任何编译错误。如果编译成功的话,那么祝贺你!


参考资料:

http://blog.csdn.net/zuolj/article/details/53168248

http://blog.csdn.net/github_30605157/article/details/60468727
内容来自用户分享和网络整理,不保证内容的准确性,如有侵权内容,可联系管理员处理 点击这里给我发消息
标签:  C++ activemq centos