Chinaunix首页 | 论坛 | 博客

justjust.blog.chinaunix.net

首页　| 　博文目录　| 　关于我

justscu

  • 博客访问： 155372
  • 博文数量： 22
  • 博客积分： 425
  • 博客等级： 下士
  • 技术积分： 350
  • 用 户 组： 普通用户
  • 注册时间： 2012-05-15 09:43
个人简介

专注服务器设计、开发； https://github.com/justscu；

文章分类

全部博文（22）

文章存档

2014年（10）

2013年（2）

2012年（10）

我的朋友
最近访客
推荐博文
相关博文
消息队列ActiveMQ

分类： 服务器与存储

2014-01-22 14:59:35

参考文档:
官网（apache.org/）
开源中国社区（）
专门介绍原理（http://blog.csdn.net/kongjing0815/article/details/8177459)
基础知识
1. 基本信息：
1. 下载activeMQ，（apache.org/activemq-570-release.html）
2. 这里以apache-activemq-5.7.0-bin.tar.gz为例。
3. 下载后，将其拷贝到/home/ll/commlib/下， 执行tar -zxvf apache-activemq-5.7.0-bin.tar.gz
4. cd apache-activemq-5.7.0，可以看见有个README.txt文件。读读有好处。
5. cd ./bin，可以看见一些可执行的脚本。运行./activemq start，启动activemq.
6. activemq默认端口为61616，用netstat -ano | grep 61616，看看该端口是否处于监听状态。
7. ./activemq stop，关闭.
8. 在apache-activemq-5.7.0/conf/目录下，有对MQ进行配置的文件。
9. 监控: in/topics.jsp

2. 性能
官方提供数据：2.1W-2.2W/second，64bit双核CPU，SUSE系统。

3. 基础知识
（1）点对点消息(queue)
        每个消息只能有一个消费者(无论消费者在消息产生的时候是否处于运行状态，都可以接收消息)。
（2）sub/pub消息(topic)
        每个消息，可以有多个消费者。消费者只能得到订阅后，生产者发布的消息。
JMS 规范允许客户创建持久订阅，持久订阅允许消费者消费它在未处于激活状态时发送的消息。
（3）消息的生产者(producer)
            由会话创建的一个对象，将消息发送到目的地。
        消息的消费者(consumer)
            (1)同步消费(receive阻塞式的接收)；(2)异步消费(设置回调函数)
（4）传输方式
        VM transport: 在VM内部通信，避免了网络传输的开销。不是采用socket通信，而是采用直接调用的方法。
        TCP transport: 客户端通过TCP的方式连接到broker。
        Failover transport: 是一种重新建立连接的机制，可靠的传输。
failover transport，允许重连的机制，建立可靠的传输;
master-slave：消息被复制到slave-broker。当master-broker遇到故障时，slave-broker不丢失任何消息。
设置randomize="false"，可以让客户总是尝试连接master broker(slave broker不会接收任何连接，直到它成为master broker).
        Discovery transport:  可靠的transport.
（5）持久化存储（persistence）
        apache.org/persistence.html
        AMQ Message Store: messages都是被保存在data logs中的，同时被reference store进行索引以提高存取速度。当某个data logs中的消息都被消费了，该data log文件就会被标记，以便在下一轮清理中被归档或删除。
        Kaha Persistence: 在 Kaha 中,数据被追加到 data logs 中。当不再需要log文件中的数据的时候，log 文件会被丢弃。
        JDBC Persistence: 将消息保存在数据库中。
（6）集群模式（clustering）
        master-slave，主从模式。消息从Master Broker复制到Slave Broker。master-broker不需要特殊配置，但slave-broker需要特殊配置

4. 一些特性
（1）当多个consumer同时从Queue中取消息时，要注意消息的顺序。Message Groups特性保证了具有相同GroupID的消息被发送给同一个consumer.
（2）JMS Selectors，可以按照关键字对消息进行过滤，某个consumer只接收关心的消息。
（3）Pending Message Limit Strategy。每个consumer都会设置一定的缓存来接收消息，当consumer的消息满时，broker不会再向consumer分发消息。但这时broker的消息就会累积，会将消息存磁盘（会带来性能问题）。consumer的一种策略是新的消息来时，将旧的消息丢掉，这样就broker就不会积累消息了。
（4）AMQ支持同步发送(sync)和异步发送(async)。async的性能更高，但可能会少量丢失消息。
（5）Strict Order Dispatch Policy，可以保证多个consumer按照同样的顺序来接收多个消息。
（6）AMQ可以采用批量确认（Optimized Acknowledgement）的方式提高效率。
（7）使用Producer Flow Control来控制producer消息的发送速度。可以限制producer在收到broker确认前，能够发送的最大字节数。

CMS (stands for C++ Messaging Service)
  1. activeMQ作为broker，可以支持多种语言的连接（如java, c++, c等）。其中，C++的地址为（apache.org/cms/）;开发接口API文档地址（apache.org/cms/api_docs/activemqcpp-3.6.0/html/index.html）。假设使用的版本为activemq-cpp-library-3.5.0.
  2. activemq-cpp-library-3.5.0的编译是比较麻烦的，依赖于APR, APR-Util, CPPUnit, OpenSSL等，可以参考README.txt文件。
  3. 开发模型图
4. 包装后的接口代码
点击(此处)折叠或打开   mq_api.h
  1. #pragma once

  3. #include <decaf/lang/Thread.h>
  4. #include <decaf/lang/Runnable.h>
  5. #include <decaf/util/concurrent/CountDownLatch.h>
  6. #include <activemq/core/ActiveMQConnectionFactory.h>
  7. #include <activemq/core/ActiveMQConnection.h>
  8. #include <activemq/transport/DefaultTransportListener.h>
  9. #include <activemq/library/ActiveMQCPP.h>
  10. #include <decaf/lang/Integer.h>
  11. #include <activemq/util/Config.h>
  12. #include <decaf/util/Date.h>
  13. #include <cms/Connection.h>
  14. #include <cms/Session.h>
  15. #include <cms/TextMessage.h>
  16. #include <cms/BytesMessage.h>
  17. #include <cms/MapMessage.h>
  18. #include <cms/ExceptionListener.h>
  19. #include <cms/MessageListener.h>

  21. #include <string>
  22. #include <vector>
  23. #include <stdlib.h>
  24. #include <stdio.h>
  25. #include <iostream>

  27. using namespace activemq;
  28. using namespace activemq::core;
  29. using namespace activemq::transport;
  30. using namespace decaf::lang;
  31. using namespace decaf::util;
  32. using namespace decaf::util::concurrent;
  33. using namespace cms;
  34. using namespace std;

  36. typedef void (*MQCbFunc)(string dest_name, unsigned char *buffer,
  37.         size_t buflen);

  39. class ActiveMQAPI: public ExceptionListener, public MessageListener, public DefaultTransportListener
  40. {
  41. public:
  42.     ActiveMQAPI(const string& brokerURI, MQCbFunc pCallBack, const string& clientID = "") :
  43.         connection(NULL), session(NULL), pCallBack(pCallBack), brokerURI(brokerURI), clientID(clientID)
  44.     {

  46.     }
  47.     virtual ~ActiveMQAPI() throw ();
  48.     void close();
  49.     int Init();

  51.     /**
  52.      *@param dest: topic/queue name
  53.      *@param name: if PersistentMode=true, name is the durable sub name
  54.      */
  55.     int Subscribe(const string& dest, bool useTopic = true, const string& name =
  56.      "", bool PersistentMode = false);
  57.     /**
  58.      @priority: 0-9,
  59.      @timeToLive:The time to live value for this message in milliseconds. 0, never expire;
  60.      */
  61.     int PubMsg(const string& dest, const string& msg, bool useTopic = true,
  62.      bool PersistentMode = false, int priority = 4,
  63.      long long timeToLive = 0);

  65.     int UnSubscribe(const string& dest, const string& name = "");

  67.     virtual void onMessage(const cms::Message* message) throw ();
  68.     virtual void onException(const CMSException& ex AMQCPP_UNUSED);
  69.     virtual void transportInterrupted();
  70.     virtual void transportResumed();

  72.     int GetDestCount()
  73.     {
  74.         return (int) dest_names.size();
  75.     }
  76. private:
  77.     void cleanup();

  79. private:
  80.     Connection* connection;
  81.     Session* session;
  82.     vector<string> dest_names;
  83.     map<string, MessageConsumer*> consumermap; // 消费者容器
  84.     map<string, MessageProducer*> producermap; // 生产者容器
  85.     map<Destination*, string> destmap;

  87.     MQCbFunc pCallBack;

  89.     const string brokerURI;
  90.     const string clientID;
  91. };
点击(此处)折叠或打开   mq_api.cpp
  1. #include "mq_api.h"

  3. void ActiveMQAPI::close() {
  4.     cleanup();
  5. }

  7. ActiveMQAPI::~ActiveMQAPI() throw () {
  8.     cleanup();
  9. }

  11. // 初始化
  12. int ActiveMQAPI::Init()
  13. {
  14.     try
  15.     {
  16.         ActiveMQConnectionFactory* connectionFactory =
  17.          new ActiveMQConnectionFactory(brokerURI);
  18.         //create connection
  19.         connection = connectionFactory->createConnection();
  20.         if (clientID.size() > 0)
  21.             connection->setClientID(clientID);
  22.         delete connectionFactory;
  23.         connectionFactory = NULL;

  25.         ActiveMQConnection* amqConnection =
  26.          dynamic_cast<ActiveMQConnection*> (connection);
  27.         if (amqConnection != NULL)
  28.         {
  29.             amqConnection->addTransportListener(this);
  30.         }

  32.         connection->start();
  33.         connection->setExceptionListener(this);

  35.         //create session
  36.         session = connection->createSession(Session::AUTO_ACKNOWLEDGE); // 自动应答
  37.     } catch (CMSException& e)
  38.     {
  39.         e.printStackTrace();
  40.         return -1;
  41.     }

  43.     return 0;
  44. }

  46. // 去broker订阅.
  47. // @dest为 topic或queue的名字
  48. // @useTopic = true，订阅的为topic；否则为queue.
  49. int ActiveMQAPI::Subscribe(const string& dest, bool useTopic,
  50.         const string& name, bool PersistentMode)
  51. {
  52.     if (session == NULL)
  53.         return -1;

  55.     Destination* destination;
  56.     MessageConsumer *consumer = NULL;

  58.     map<string, MessageConsumer*>::iterator it;
  59.     it = consumermap.find(dest);
  60.     if (it != consumermap.end())
  61.         return 0;
  62.     else
  63.     {
  64.         if (useTopic)
  65.             destination = session->createTopic(dest);
  66.         else
  67.             destination = session->createQueue(dest);

  69.         dest_names.push_back(dest);
  70.         destmap.insert(pair<Destination*, string> (destination, dest));
  71.         if (useTopic && PersistentMode)
  72.         {
  73.             if (name.size() == 0)
  74.                 return -1;
  75.             // 持久订阅
  76.             consumer = session->createDurableConsumer(
  77.              (cms::Topic*) destination, name, "");
  78.         }
  79.         else
  80.         {
  81.             consumer = session->createConsumer(destination);
  82.         }
  83.         consumermap.insert(pair<string, MessageConsumer*> (dest, consumer));
  84.         consumer->setMessageListener(this);
  85.     }

  87.     return 0;
  88. }

  90. // 发布
  91. int ActiveMQAPI::PubMsg(const string& dest, const string& msg, bool useTopic,
  92.         bool PersistentMode, int priority, long long timeToLive)
  93. {
  94.     if (msg.size() == 0 || dest.size() == 0)
  95.         return -1;
  96.     if (session == NULL)
  97.         return -1;

  99.     Destination* destination = NULL;
  100.     MessageProducer* producer = NULL;

  102.     map<string, MessageProducer*>::iterator it;
  103.     it = producermap.find(dest);
  104.     if (it != producermap.end())
  105.     {
  106.         producer = it->second;
  107.     }
  108.     else
  109.     {
  110.         if (useTopic)
  111.             destination = session->createTopic(dest);
  112.         else
  113.             destination = session->createQueue(dest);
  114.         producer = session->createProducer(destination);

  116.         if (PersistentMode)
  117.             producer->setDeliveryMode(DeliveryMode::PERSISTENT);
  118.         else
  119.             producer->setDeliveryMode(DeliveryMode::NON_PERSISTENT);

  121.         producermap.insert(pair<string, MessageProducer*> (dest, producer));
  122.     }

  124.     BytesMessage* message = session->createBytesMessage((unsigned char*) msg.c_str(), msg.size());
  125.     // producer->send(message);
  126.     producer->send(message, (PersistentMode == true) ? DeliveryMode::PERSISTENT
  127.      : DeliveryMode::NON_PERSISTENT, priority, timeToLive);

  129.     delete message;
  130.     message = NULL;
  131.     return 0;
  132. }

  134. // 取消订阅
  135. int ActiveMQAPI::UnSubscribe(const string& dest, const string& name) {
  136.     map<string, MessageConsumer*>::iterator it_c = consumermap.find(dest);
  137.     MessageConsumer* consumer = NULL;
  138.     if (it_c != consumermap.end())
  139.     {
  140.         consumer = it_c->second;
  141.         consumermap.erase(it_c);
  142.         delete consumer;
  143.         consumer = NULL;
  144.     }

  146.     map<Destination*, string>::iterator it;
  147.     Destination* destination = NULL;
  148.     for (it = destmap.begin(); it != destmap.end();)
  149.     {
  150.         Destination* k = it->first;
  151.         const string &v = it->second;
  152.         it++;
  153.         if (dest.compare(v) == 0)
  154.         {
  155.             destination = k;
  156.             destmap.erase(k);

  158.             try
  159.             {
  160.                 if (NULL == destination)
  161.                 {
  162.                     printf(
  163.                      "ActiveMQAPI(UnSubscribe)[%s]: destination[%p] is NULL",
  164.                      dest.c_str(), destination);
  165.                     return -1;
  166.                 }
  167.                 ActiveMQConnection* amqConnection =
  168.                  dynamic_cast<ActiveMQConnection*> (connection);
  169.                 amqConnection->destroyDestination(destination);
  170.             } catch (std::exception& e)
  171.             {
  172.                 printf("ActiveMQAPI(UnSubscribe)[%s]: exception[%s]",
  173.                  dest.c_str(), e.what());
  174.                 return -1;
  175.             }

  177.             delete destination;
  178.             destination = NULL;

  180.             break;
  181.         }
  182.     }

  184.     if (name.size() > 0)
  185.     {
  186.         if (session == NULL)
  187.             return -1;
  188.         try
  189.         {
  190.             session->unsubscribe(name);
  191.         } catch (CMSException& e)
  192.         {
  193.             e.printStackTrace();
  194.             return -1;
  195.         }
  196.     }
  197.     return 0;

  199. }

  201. // 消息来时的回调函数
  202. void ActiveMQAPI::onMessage(const Message* message) throw ()
  203. {
  204.     try
  205.     {
  206.         const BytesMessage* msg = dynamic_cast<const BytesMessage*> (message);
  207.         unsigned char *buffer = NULL;
  208.         size_t buflen = 0;
  209.         string destname;

  211.         if (msg != NULL)
  212.         {
  213.             buffer = msg->getBodyBytes();
  214.             buflen = msg->getBodyLength();
  215.             map<Destination*, string>::iterator it;
  216.             for (it = destmap.begin(); it != destmap.end(); it++)
  217.             {
  218.                 if (msg->getCMSDestination()->equals(*(it->first)))
  219.                 {
  220.                     destname = it->second;
  221.                 }
  222.             }
  223.         }
  224.         else
  225.         {
  226.             printf("the msg is NULL");
  227.             return;
  228.         }

  230.         pCallBack(destname, buffer, buflen);
  231.         if (NULL != buffer)
  232.         {
  233.             delete[] buffer;
  234.             buffer = NULL;
  235.         }

  237.     }
  238.     catch (CMSException& e)
  239.     {
  240.         e.printStackTrace();
  241.     }
  242. }

  244. void ActiveMQAPI::onException(const CMSException& ex AMQCPP_UNUSED)
  245. {
  246.     printf("ActiveMQAPI: CMS Exception occurred[%s]. Shutting down client.",
  247.      ex.what());

  249.     // exit(1);
  250. }

  252. void ActiveMQAPI::transportInterrupted()
  253. {
  254.     printf("ActiveMQAPI: The Connection's Transport has been Interrupted.");
  255. }

  257. void ActiveMQAPI::transportResumed()
  258. {
  259.     printf("ActiveMQAPI: The Connection's Transport has been Restored.");
  260. }

  262. void ActiveMQAPI::cleanup()
  263. {
  264.     map<Destination*, string>::iterator it;
  265.     for (it = destmap.begin(); it != destmap.end();)
  266.     {
  267.         Destination* k = it->first;
  268.         it++;
  269.         destmap.erase(k);
  270.         delete k;
  271.         k = NULL;
  272.     }

  274.     map<string, MessageConsumer*>::iterator it_consumer;
  275.     for (it_consumer = consumermap.begin(); it_consumer != consumermap.end();)
  276.     {
  277.         const string &k = it_consumer->first;
  278.         const MessageConsumer* v = it_consumer->second;
  279.         it_consumer++;
  280.         if (v != NULL)
  281.         {
  282.             delete v;
  283.             v = NULL;
  284.             consumermap.erase(k);
  285.         }
  286.     }

  288.     if (session != NULL)
  289.     {
  290.         session->close();
  291.         delete session;
  292.         session = NULL;
  293.     }
  294.     if (connection != NULL)
  295.     {
  296.         connection->close();
  297.         delete connection;
  298.         connection = NULL;
  299.     }
  300. }

5. 测试代码
点击(此处)折叠或打开   订阅topic
  1. void MQCallBack(string dest_name, unsigned char *buffer,
  2.         size_t buflen)
  3. {
  4.     printf("topic[%s] buffer[%s] \n", dest_name.c_str(), string((char*)buffer, buflen).c_str());
  5. }

  7. int main()
  8. {
  9.     activemq::library::ActiveMQCPP::initializeLibrary();
  10.     //
  11.     ActiveMQAPI* mq = new ActiveMQAPI("failover:(tcp://XX:61616,tcp://XX:61616)?randomize=false", MQCallBack);
  12.     if(NULL == mq)
  13.     {
  14.         printf("new activeMQ error.");
  15.         return 0;
  16.     }

  18.     if(0 != mq->Init())
  19.     {
  20.         printf("init activeMQ failed.");
  21.         delete mq;
  22.         mq = NULL;
  23.         return 0;
  24.     }

  26.     mq->Subscribe("bond.mq", true);

  28.     while(true)
  29.     {
  30.         sleep(5);
  31.     }

  33.     mq->UnSubscribe("bond.mq");
  34.     delete mq;
  35.     mq = NULL;
  36.     activemq::library::ActiveMQCPP::shutdownLibrary();
  37.     return 0;
  38. }
        订阅后，可以通过 MQ提供的监控平台来查看是否订阅成功。在（）主页的 “Connections”下面，通过IP找到运行订阅程序的主机，然后再点击进去，看是否有这个topic。

点击(此处)折叠或打开   发布topic

  1. int main()
  2. {
  3.     activemq::library::ActiveMQCPP::initializeLibrary();
  4.     //
  5.     ActiveMQAPI* mq = new ActiveMQAPI("failover:(tcp://XX:61616,tcp://XX:61616)?randomize=false", MQCallBack);
  6.     if(NULL == mq)
  7.     {
  8.         printf("new activeMQ error.");
  9.         return 0;
  10.     }

  12.     if(0 != mq->Init())
  13.     {
  14.         printf("init activeMQ failed.");
  15.         delete mq;
  16.         mq = NULL;
  17.         return 0;
  18.     }

  20.     int iRet = mq->PubMsg("bond.mq", "the message to send to the consumers.", true, true);
  21.     delete mq;
  22.     mq = NULL;

  24.     activemq::library::ActiveMQCPP::shutdownLibrary();
  25.     return 0;
  26. }



阅读(5156) | 评论(0) | 转发(0) |
0

上一篇：用epoll实现一个异步处理的模型(2)

下一篇：Nginx之一：源文件编译、安装、测试

给主人留下些什么吧！~~
关于我们 | 关于IT168 | 联系方式 | 广告合作 | 法律声明 | 免费注册

Copyright 2001-2010 ChinaUnix.net All Rights Reserved 北京皓辰网域网络信息技术有限公司. 版权所有

感谢所有关心和支持过ChinaUnix的朋友们

16024965号-6