消息队列调研报告

<消息队列>技术调研报告摘要“消息队列网络”确保消息顺当处理的过程中扮演不同的角色。它们中有些供给路由信息以确定如何发送消息，有些保存整个网络的重要信息，而有些只是发送和接收消息。调研目的本次调研的目的通过了解activemq,rabbitmq能更牢靠，以及通过三个队列在优化前和优化后的性能比照，看哪共性能更好一些。Activemq简介ActiveMQ是Apache出品，最流行的，力量强劲的开源消息总线。ActiveMQ是一个完全支持JMS1.1和J2EE1.4标准的JMSProvider实现，尽管JMS标准出台已经是很久的事情了，但是JMS在当今的J2EE应用中间仍旧扮演着特别的地位。多种语言和协议编写客户端。语言:Java,C,C++,C#,Ruby,Perl,Python,PHP。应用协议：OpenWire,StompREST,WSNotification,XMPP,AMQP.完全支持JMS1.1和J2EE1.4标准Spring的支持，ActiveMQ可以很简洁内嵌到使用SpringSpring2.0的特性通过了常见J2EE效劳器〔如Geronimo,JBoss4,GlassFish,WebLogic)的测试，其中通过JCA1.5resourceadaptors的配置，可以让ActiveMQ可以自动的部署到任何兼容J2EE1.4商业效劳器上支持多种传送协议：in-VM,TCP,SSL,NIO,UDP,JGroups,JXTA支持通过JDBCjournal供给高速的消息长久化从设计上保证了高性能的集群，客户端-效劳器，点对点支持Ajax支持与Axis的整合可以很简洁得调用内嵌JMSprovider，进展测试。Activemq关键字目的〔Destination〕〔管道模式中，目的被称为队列，在公布-订阅模式中，目的被称为主题。队列〔Queue〕在点对点消息模式中被作为目的。它能把每个产生的消息发送给对应唯一的订阅者进展消费。例如，每条点〕被安排给特定的接收者〔点。主题〔Topic〕作为公布-订阅消息模式中的目的类型，并复制每个所产生的消息给多个订阅者进展消费。消息〔Message〕成的特定动作。持续订阅〔DurableSubscrition〕赐予客户端能访问基于唯一名称所公布的消息的消息。非持续订阅〔Non-DurableSubscription〕赐予客户端能访问只当订阅者在线时对某个主题所公布的消息。长久传输模式〔PersistantDeliveryMode〕能保证固然JMS效劳器崩溃时所传输的消息不会丧失。非长久传输模式〔Non-PersistantDeliveryMode〕供给最低保证传输方式，当JMS效劳器崩溃时无法保证所传输的消息不会丧失。〔消息〕供给者〔Producer〕作为某个组件，客户端能发送消息给JMS供给商并传输给最终目的。〔消息〕消费者〔Consumer〕作为描述某个组件被客户端用于恳求JMS供给商进展消息接收的最常用词汇牢靠性〔Reliability〕作为效劳质量Qo，例如，具有在某个时间段内传输消息的容错级别。安全性〔Secutiry〕包括支持消息认证，授权和加密项。可延展性〔Scalability〕基于总连接数，可支持的客户总数，消息处理的LAN/WAN带宽。速度〔Speed〕和厂商相关，由消息数进展定义。公布者〔Publisher〕定义为在公布-订阅模式中，客户端能通过其发送〔产生〕消息给主题。订阅者〔Subscriber〕定义为在公布-订阅模式中，客户端能通过其接收从主题发来的消息。Activemq根本操作linux下安装解压：解压：tarzxvfactivemq-x.x.x.tar.gzbin文件夹，运行：./activemqstart&，也可以只运行：./activemqconsole。验证是否安装成功在扫瞄器中输入:://xxx.xxx.xxx.xxx:8161/admin假设消灭如下界面，则说明安装成功:修改监控台的用户名和密码:jetty.xml:<beanid=“securityConstraint“class=“org.eclipse.jetty.util.security.Constraint“><propertyname=“name“value=“BASIC“/><propertyname=“roles“value=“admin“/><propertyname=“authenticate“value=“false“/></bean>把authenticate属性的值改成true即可,重启动之后，在登录监控台，就需要输入密码了，默认的用户名和密码是admin/admin,roles属性指的是登录的用户角色，这些登录的用户在perties配置。修改端口：在jetty.xml找到一下配置，修改8161即可。<propertyname=“connectors“><list><beanid=“Connector“class=“org.eclipse.jetty.server.nio.SelectChannelConnector“><propertyname=“port“value=“8161“/></bean></list></property>55Activemq消息长久化：broker中设置属性persistent=”tr(true)，同时发送的消息也应当是persitent类型的。ActiveMQ消息长久化有三种方式：AMQ、KahaDB、JDBC。5.1AMQ:AMQ是一种文件存储形式，它具有写入速度快和简洁恢复的特点。消息存储在一个个文件中，文件的默认大小为32兆，假设一条消息的大小超过了32兆，那么这个值必需设置大点。当一个存储文件中的消息已经全部被消费，那么这个文件将被标识为可删除，在下一个去除阶段，这个文件被删除。默认配置如下：<persistenceAdapter><amqPersistenceAdapterdirectory=“activemq-data“maxFileLength=“32mb“/></persistenceAdapter>5.2KahaDBKahaDB是基于文件的本地数据库储存形式，虽然没有AMQ的速度快，但是它具有强扩展AMQ短，从5.4版本之后KahaDB做为默认的长久化方式。默认配置如下：<persistenceAdapter><persistenceAdapter><kahaDBdirectory=“activemq-data“journalMaxFileLength=“32mb“/></persistenceAdapter>5.3jdbc:配置5.3jdbc:配置JDBC适配器：<persistenceAdapter><jdbcPersistenceAdapterdataSource=“#mysql-ds“createTablesOnStartup=“false“/></persistenceAdapter>dataSourcebean，createTablesOnStartup是否在启动的时候创立数据表，默认值是true，这样每次启动都会去创立数据表了，一般是第一次启动的时候设置为truefalse。MYSQLbean<beanid=“mysql-ds“class=“org.apachemons.dbcp.BasicDataSource“destroy-method=“close“><propertyname=“driverClassName“value=“com.mysql.jdbc.Driver“/><propertyname=“url“value=“jdbc:mysql://localhost/activemq?relaxAutoCommit=true“/><propertyname=“username“value=“activemq“/><propertyname=“password“value=“activemq“/><propertyname=“poolPreparedStatements“value=“true“/></bean>Activemq集群Activemq.xmlActivemq.xml关键配置参数:Broker:brokername:brokeruseJmx:是否开启jmxPersistent:是否存储累积消息Desitination:Desitination:依据不通状况来做优化我们目前使用的是queue，所以使用queuepolicy,这里的“>“是通配符，表示全部的queuememoryLimit=“100mb“表示queue100mbproducerFlowControl=“false,amq会主动掌握流量，削减消息生产。NetworkConnector：NetworkConnector：衡。SystemUsage:SystemUsage:配置了一些系统内存和硬盘容量，当系统消耗超过这些容量设置时， amq会“slowdownproduer“,这是很重要的。TransportConnector:TransportConnector:brokerbrokerbrokerJetty:配置了一个Jetty:配置了一个jetty效劳器，供给webBrokerclusters/Networksofbrokers一个常见的场景是有多个一个常见的场景是有多个JMSbrokerbroker个broker失效，那么客户会自动重连接到其它的broker。在ActiveMQ中使用failover://协议来实现这个功能。假设某个网络上有多个brokers而且客户使用静态觉察〔使用StaticTransport或FailoverTransport〕动态觉察〔使用DiscoveryTransport，那么客户可以简洁地在某个broker失效的状况下切换到其它的brokersbroker时，brokerconsumersbroker上的消息可能会因得不处处理而积压起来。目前的解决方案是使用Networkofbrokers，以便在broker之间存储转发消息。如图：如图：ActiveMQ可以实现多个mqmbrokerA和当有一条消息发送到brokerA的队列test中，有一个客户端连接到brokerB上，并且要求猎取test队列的消息时，brokerA中队列test的消息就会路由到brokerBbrokerB的消息也会路由到brokerA。1静态觉察1静态觉察配置方法：Broker1:1>备份activemq.xmlMvactivemq.xmlactivemq_cp.xml22>将activemq-static-network-broker2.xml改为activemq.xmlMvactivemq-static-network-broker2.xmlactivemq.xml33>修改brokername找到<broker“:///schema/core“xmlns=“:///schema/core“brokerName=“static-broker2“dataDirectory=“${activemq.base}“>改为<broker改为<broker“:///schema/core“xmlns=“:///schema/core“brokerName=“static-broker1“dataDirectory=“${activemq.base}“>44>修改要监听的另外一个broker找到<networkConnectors><networkConnectoruri=“static:(tcp://localhost:61616)“duplex=“true“/> </networkConnectors>改为<networkConnectors> <networkConnectoruri=“static:(tcp://22:61616,....)“ duplex=“true“/></networkConnectors>55>修改persistenceAdapter找到<persistenceAdapter>directory=“${activemq.base}/static-broker2/kahadb“</persistenceAdapter>改为<persistenceAdapter>directory=“${activemq.base}/static-broker1/kahadb“</persistenceAdapter><kahaDB/><kahaDB/>66>修改transportConnectors找到<transportConnectors>uri=“tcp://:61618“/>改为<transportConnectors>uri=“tcp://22:61616“/><transportConnectorname=“openwire“</transportConnectors><transportConnectorname=“openwire“</transportConnectors>Broker2:Broker2:1>备份activemq.xmlMvactivemq.xmlactivemq_cp.xml22>将activemq-static-network-broker2.xml改为activemq.xmlMvactivemq-static-network-broker2.xmlactivemq.xml33>修改brokername找到<broker“:///schema/core“xmlns=“:///schema/core“brokerName=“static-broker2“dataDirectory=“${activemq.base}“>改为<broker “:///schema/core“xmlns=“:///schema/core“brokerName=“static-broker2brokerName=“static-broker2“dataDirectory=“${activemq.base}“>44>修改要监听的另外一个broker找到<networkConnectors><networkConnectoruri=“static:(tcp://localhost:61616)“duplex=“true“/> </networkConnectors>改为<networkConnectors> <networkConnectoruri=“static:(tcp://21:61616,......)“ duplex=“true“/></networkConnectors>55>修改persistenceAdapter找到<persistenceAdapter>directory=“${activemq.base}/static-broker2/kahadb“</persistenceAdapter>改为<persistenceAdapter>directory=“${activemq.base}/static-broker2/kahadb“</persistenceAdapter><kahaDB/><kahaDB/>66>修改transportConnectors找到<transportConnectors>uri=“tcp://:61618“/>改为<transportConnectors>uri=“tcp://21:61616“/><transportConnectorname=“openwire“</transportConnectors><transportConnectorname=“openwire“</transportConnectors>22动态觉察:动态路由配置，每个mq都需要配置如下：<networkConnectors><networkConnectoruri=“multicast://default“/></networkConnectors><transportConnectors><transportConnector name=“openwire“ uri=“tcp://:61618“discoveryUri=“multicast://default“/></transportConnectors>MasterMasterSlave简介：在一个网络内运行多个brokers或者standalonebrokers时存在一个问题，这就是消息在物理上只被一个broker这就是消息在物理上只被一个broker持有，因此当某个broker失效，那么你只能等待直到它重启后，这个broker上的消息才能够被连续发送〔假设没有设置长久化，那么在这种状况下，消息将会丧失。MasterSlave背后的想法是，消息被复制到slavebroker，因此即使masterbroker遇到了像硬件故障之类的错误，你也可以马上切换到slavebroker而不丧失任何消息。主要分为三类：PureMasterSlave/SharedFileSystemMasterSlave/JDBCMasterSlavePureMasterSlave一个master只对应一个slavePureMasterSlave具有以下限制：只能有一个slavebrokermasterbroker。在masterbroker而导致slavebrokermaster，之前的masterbroker只有在当前的masterbroker〔原slavebroker〕停顿后才能重生效。Masterbrokerslavebroker最安全的恢复masterbroker的方式是人工处理。首先要停顿slavebroker〔这意味着全部的客户也要停顿。然后把slavebroker的数据名目中全部的数据拷贝到masterbroker。然后重启masterbrokerslavebroker。Masterbroker不需要特别的配置。SlavebrokerXmlshutdownOnMasterFailure=“false“>2....3.<transportConnectors>4.<transportConnectoruri=“tcp://22:61616“/></transportConnectors></broker>其中的masterConnectorURI用于指向masterbroker，shutdownOnMasterFailureslavebrokermasterbroker候是否需要停顿。此外，也可以使用如下配置：Xml1.<brokerbrokerName=“slave“useJmx=“false“deleteAllMessagesOnStartup=“true““:///config/1.0“xmlns=“:///config/1.0“>2....<services><masterConnectorremoteURI=“tcp://21:61616“userName=“user“password=“password“/></services></broker>数据库共享方式数据库共享方式这种方式的主备承受数据库做消息的长久化，支持多个Slave，全部broker长久化数据源配brokerbroker都成为slave并且等待猎取锁，当master挂了之后，其中的一个slave将会马上获得数据库锁成为master，重启mastermasterslaveslave来恢复。由于承受的是数据库做为长久化，它的性能是有限的.特点：1配置上，不存在MasterSlave，全部Broder的配置根本是一样的2多个共享数据源的Broker构成JDBCMasterSlave给每个Broker取个名字首先抢到资源〔数据库锁〕的Broker成为Maseter5其他Broker保持预备状态6一旦Master崩溃，其他Broker尝试抢占资源，最终只有一台抢到，它马上成为Master7之前的Master即使重启成功，也只能作为Slave等待配置方式：首先修改<brokerbrokerName=“jdbc_broker01“…<!--配置长久适配器--><persistenceAdapter><jdbcPersistenceAdapterdataDirectory=“${activemq.base}/data“dataSource=“#mysql-ds“/></persistenceAdapter><transportConnectorname=“openwire“uri=“tcp://:61616“/><!--配置数据源：留意是在broker标记之外-->…</broker><beanid=“mysql-ds“class=“org.apachemons.dbcp.BasicDataSource“destroy-method=“close“><propertyname=“driverClassName“value=“com.mysql.jdbc.Driver“/><propertyname=“url“value=“jdbc:mysql://localhost/test?relaxAutoCommit=true“/><propertyname=“username“value=“root“/><propertyname=“password“value=“root“/><propertyname=“maxActive“value=“200“/><propertyname=“poolPreparedStatements“value=“true“/></bean>其他的broker配置同上，只需要修改名称即可。rabbitmq简介测试用例〔Activemq〕测试目的通过对activemq的优化，对一样条件下的数据进展测试，推断进展什么样的优化可以对activemq的吞吐量有提高。测试用例测试方法：Producer测试用例测试方法：Producer100000条数据测试过程：1启动Producer2记录开头插入数据的时间3全部数据插入完毕后，记录时间测试结果：队列名称测试结果：队列名称Producer1000006279ms测试用例测试方法：Consumer100000测试用例测试方法：Consumer100000条数据测试过程：1启动Consumer2记录开头猎取数据的时间3全部数据猎取完毕后，记录时间测试结果：测试结果：队列名称Consumer10000024294ms测试用例测试方法：生产者往消息队列中放入100000条数据测试用例测试方法：生产者往消息队列中放入100000条数据消费者从消息队列中猎取消息测试过程：1启动Consumer,Producer2记录开头发送，猎取数据的时间3全部数据猎取完毕后，记录时间测试结果：测试结果：1000008346ms接收消息39157ms测试用例测试方法：生产者往消息队列中放入100000条数据消费者从消息队列中猎取消息测试过程：1启动测试用例测试方法：生产者往消息队列中放入100000条数据消费者从消息队列中猎取消息测试过程：1启动Consumer,Producer2记录开头发送，猎取数据的时间3全部数据猎取完毕后，记录时间测试结果：测试结果：10000011994ms接收消息15678ms测试用例测试方法：生产者往消息队列中放入100000条数据消费者从消息队列中猎取消息测试过程：1启动Consumer,Producer测试用例测试方法：生产者往消息队列中放入100000条数据消费者从消息队列中猎取消息测试过程：1启动Consumer,Producer2记录开头发送，猎取数据的时间3全部数据猎取完毕后，记录时间测试结果：测试结果：10000011965ms接收消息11939ms测试用例测试方法：1修改<policyEntryqueue=“>“producerFlowControl=“false“memoryLimit=“1024mb“>测试用例测试方法：1修改<policyEntryqueue=“>“producerFlowControl=“false“memoryLimit