ActiveMQ 环境搭建与测试(C++)

作者: 贰爷 | 来源:发表于2020-04-12 15:18 被阅读0次

一、下载安装jdk

jdk官网下载地址：https://www.oracle.com/technetwork/java/javase/downloads/jdk8-downloads-2133151.html

1、解压缩到安装目录下

２、编辑配置文件，配置环境变量

终端执行命令：vim /etc/profile
添加如下内容：JAVA_HOME根据实际目录来:

export JAVA_HOME=/home/ss/software/jdk1.8.0_191
export JRE_HOME=${JAVA_HOME}/jre
export CLASSPATH=.:${JAVA_HOME}/lib:${JRE_HOME}/lib
export PATH=${JAVA_HOME}/bin:$PATH

终端执行命令：
source /etc/profile
3、查看安装情况
java -version

二、下载安装ActiveMQ

ActiveMQ官网下载地址：http://activemq.apache.org/download.html

1、解压缩，进入根目录下，进入bin文件，在文件里面打开终端，输入如下命令启动ActiveMQ

./activemq start

停止则输入：

./activemq stop

2、查看是否启动：

ps -elf|grep active

3、监控

ActiveMQ默认启动时，启动了内置的jetty服务器，提供一个用于监控ActiveMQ的admin应用。

访问链接形式：http://127.0.0.1:8161/admin/

在浏览器中访问上面的链接（注意：ip修改成activeMq服务端的ip地址），出现登录界面，输入用户名和密码会进入activeMq管理网页页面（用户和密码都是admin），可以从这个网页监控端查看队列名称、生产的消息数目以及消费情况以及是否有消费者正在等待消息的到来。

在这里插入图片描述

三、下载安装依赖库及ActiveMQ CPP

参考：https://blog.csdn.net/lgh1700/article/details/51055784

打开http://activemq.apache.org/cms/building.html页面，这里介绍了cms build时用到的依赖库。

1、cppunit

cppunit下载页面：
https://sourceforge.net/projects/cppunit/files/cppunit/
tar解压后，进入目录，编译三部曲，configure、make、make install（install需要root权限）：

./configure --prefix=/usr/local/cppunit/
make && make install

执行完后在/usr/local/cppunit/目录下可以看到头文件和库文件。
如果在编译过程出现如下错误：

../../src/cppunit/.libs/libcppunit.so  -Wl,--rpath -Wl,/usr/local/cppunit//lib
../../src/cppunit/.libs/libcppunit.so: undefined reference to `dlopen'
../../src/cppunit/.libs/libcppunit.so: undefined reference to `dlclose'
../../src/cppunit/.libs/libcppunit.so: undefined reference to `dlsym'

命令改为

./configure LDFLAGS='-ldl' --prefix=/usr/local/cppunit/
make && make install

或在Ｍakefile中加入

DFLAGS=-lpthread -lrt -ldl

2、apr

apr的全称为Apache Portable Runtime（Apache可移植运行时），Apache旗下有很多开源软件。
apr介绍页面：
http://apr.apache.org/download.cgi
apr下载地址（与以下的apr-util和apr-iconv下载页面一致，下载时注意区分名称）：
http://mirrors.hust.edu.cn/apache/apr/
同上，解压进入目录，三部曲：

./configure --prefix=/usr/local/apr/
make && make install

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

3、apr-util

apr-util下载地址：
http://mirrors.hust.edu.cn/apache/apr/
解压进入目录编译：

./configure --prefix=/usr/local/aprutil --with-apr=/usr/local/apr/
make && make install

4、apr-iconv

apr-iconv下载地址：
http://mirrors.hust.edu.cn/apache//apr/
解压编译：

./configure --prefix=/usr/local/apr-iconv/ --with-apr=/usr/local/apr/
make && make install

5、openssl

openssl下载地址：
https://www.openssl.org/source/

解压编译：

./config --prefix=/usr/local/openssl/
make && make install

若出现报错
cms.pod around line 457: Expected text after =item, not a number
在root权限下，执行

rm -f /usr/bin/pod2man

然后重新

make install

6、ActiveMQ-CPP

ActiveMQ-CPP下载地址：
http://activemq.apache.org/cms/download.html
解压编译：

./configure --prefix=/usr/local/ActiveMQ-CPP --with-apr=/usr/local/apr/ --with-apr-util=/usr/local/aprutil --with-cppunit=/usr/local/cppunit --with-openssl=/usr/local/openssl
make && make install

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

四、测试

采用该博客的代码进行测试(代码已搬运到文末)：
https://blog.csdn.net/chenxun_2010/article/details/52709277

1、编译生成可执行文件：

g++ producer.cpp -o send -I/usr/local/ActiveMQ-CPP/include/activemq-cpp-3.9.4 -I/usr/local/apr/include/apr-1 -L/usr/local/ActiveMQ-CPP/lib -lactivemq-cpp

g++ consumer.cpp -o receive -I/usr/local/ActiveMQ-CPP/include/activemq-cpp-3.9.4 -I/usr/local/apr/include/apr-1 -L/usr/local/ActiveMQ-CPP/lib -lactivemq-cpp

2、运行

./send
./receive

3、出错：找不到依赖库

把常用的依赖库添加到动态路径里面重新运行即可：

cat /etc/ld.so.conf
echo "/usr/local/lib" >> /etc/ld.so.conf
echo "/usr/local/apr/lib" >> /etc/ld.so.conf
echo "/usr/local/ActiveMQ-CPP/lib" >> /etc/ld.so.conf
ldconfig

代码如下：producer.cpp和consumer.cpp

producer.cpp:

#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 {
        //1、创建工厂连接对象，需要制定ip和端口号
            auto_ptr<ActiveMQConnectionFactory> connectionFactory(new ActiveMQConnectionFactory( brokerURI ) );

           try{
        //2、使用连接工厂创建一个连接对象
                connection = connectionFactory->createConnection();
        //3、开启连接
                connection->start();
            } catch( CMSException& e ) {
                e.printStackTrace();
                throw e;
            }
        //4、使用连接对象创建会话（session）对象 
            if( clientAck ) 
            {
                session = connection->createSession( Session::CLIENT_ACKNOWLEDGE );
            } else 
            {
                session = connection->createSession( Session::AUTO_ACKNOWLEDGE );
            }
         //5、使用会话对象创建目标对象，包含queue和topic（一对一和一对多）
            if( useTopic ) 
            {
                destination = session->createTopic( destURI );
            } else 
            {
                destination = session->createQueue( destURI );
            }
        //6、使用会话对象创建生产者对象
            producer = session->createProducer( destination );
            producer->setDeliveryMode( DeliveryMode::NON_PERSISTENT );

            string threadIdStr = Long::toString( Thread::currentThread()->getId() );
            string text = (string)"Hello world! from thread " + threadIdStr;

            for( unsigned int ix=0; ix<numMessages; ++ix )
            {
        //7、使用会话对象创建一个消息对象
                TextMessage* message = session->createTextMessage( text );
                message->setIntProperty( "Integer", ix );
                printf( "Sent message #%d from thread %s\n", ix+1, threadIdStr.c_str() );
        //8、发送消息
                producer->send( message );
                delete message;
            }

        }catch ( CMSException& e ) {
            e.printStackTrace();
        }
    }

private:
        //9、关闭资源
    void cleanup(){
        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;

        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 , char* argv[]) 
{

    activemq::library::ActiveMQCPP::initializeLibrary();
    std::cout << "=====================================================\n";
    std::cout << "Starting produce message:" << std::endl;
    std::cout << "-----------------------------------------------------\n";

    std::string brokerURI ="failover://(tcp://127.0.0.1:61616)";
    unsigned int numMessages = 2000;
    std::string destURI = "test.chen";

    bool useTopics = false;
    SimpleProducer producer( brokerURI, numMessages, destURI, useTopics );
    producer.run();
    producer.close();

    std::cout << "-----------------------------------------------------\n";
    std::cout << "Finished test" << std::endl;
    std::cout << "=====================================================\n";

    activemq::library::ActiveMQCPP::shutdownLibrary();
}

consumer.cpp:

#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 {
    //1、创建工厂连接对象，需要制定ip和端口号
            ActiveMQConnectionFactory* connectionFactory = new ActiveMQConnectionFactory( brokerURI );
    //2、使用连接工厂创建一个连接对象
            connection = connectionFactory->createConnection();
            delete connectionFactory;

            ActiveMQConnection* amqConnection = dynamic_cast<ActiveMQConnection*>( connection );
            if( amqConnection != NULL ) 
            {
                amqConnection->addTransportListener( this );
            }
    //3、开启连接
            connection->start();
            connection->setExceptionListener(this);
    //4、使用连接对象创建会话（session）对象       
            if( clientAck ) {
                session = connection->createSession( Session::CLIENT_ACKNOWLEDGE );
            } else {
                session = connection->createSession( Session::AUTO_ACKNOWLEDGE );
            }
    //5、使用会话对象创建目标对象，包含queue和topic（一对一和一对多）
            if( useTopic ) {
                destination = session->createTopic( destURI );
            } else {
                destination = session->createQueue( destURI );
            }
    //6、使用会话对象创建生产者对象
            consumer = session->createConsumer( destination );
    //7、向consumer对象中设置一个messageListener对象，用来接收消息
            consumer->setMessageListener( this );

        } catch (CMSException& e) {
            e.printStackTrace();
        }
    }
    //8、程序等待接收用户消息
    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();
        }
    }

    virtual void onException( const CMSException& ex AMQCPP_UNUSED ) {
        printf("CMS Exception occurred.  Shutting down client.\n");
        exit(1);
    }

    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:
     //9、关闭资源
    void cleanup(){
        try{
            if( destination != NULL ) delete destination;
        }catch (CMSException& e) {}
        destination = NULL;

        try{
            if( consumer != NULL ) delete consumer;
        }catch (CMSException& e) {}
        consumer = NULL;

        try{
            if( session != NULL ) session->close();
            if( connection != NULL ) connection->close();
        }catch (CMSException& e) {}

        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, char* argv[]) {

    activemq::library::ActiveMQCPP::initializeLibrary();

    std::cout << "=====================================================\n";
    std::cout << "Starting the example:" << std::endl;
    std::cout << "-----------------------------------------------------\n";

    std::string brokerURI = "failover:(tcp://127.0.0.1:61616)";

    std::string destURI = "test.chen"; 
    bool useTopics = false;
    bool clientAck = false;
    SimpleAsyncConsumer consumer( brokerURI, destURI, useTopics, clientAck );
    consumer.runConsumer();
    std::cout << "Press 'q' to quit" << std::endl;
    while( std::cin.get() != 'q') {}
    consumer.close();

    std::cout << "-----------------------------------------------------\n";
    std::cout << "Finished with the example." << std::endl;
    std::cout << "=====================================================\n";

    activemq::library::ActiveMQCPP::shutdownLibrary();
}
}

参考链接：
http://blog.sina.com.cn/s/blog_a459dcf501017ooh.html
https://blog.csdn.net/liuyuanq123/article/details/79109218