ActiveMQ性能调优

转自 ： http://setting.iteye.com/blog/1484498

转自： http://www.langyuweb.com/a/jishuzhongxin/lanmu1/2011/0823/4225.html

Chapter 1.

Introduction to Performance Tuning

这章节首要概括性的介绍一些用于优化持久broker和非持久broker的一些技巧。

Limiting factors

谈及如何提拔机能之前，我们必须知道是那些身分影响了机能，大体首要包含以下三个方面：

·       消息从磁盘中写入和读取（仅限与持久消息）

·       消息的编组和基于收集的传递

·       基于多线程的高低文切换

终极，我们会按照以上身分来调优。

Non-persistent and persistent brokers

非持久的broker调优设置和持久broker的调优设置稍微有一点不合，然则，大项目组的调优技巧将在该章用于持久和非持久broker。

System Environment

Overview

在评论辩论如何优化机能之前，必须重视一下体系景象的设备，因为它也能引起一些影响对于机能。

Disk speed

对于持久的broker来说，磁盘的转速也是一个很首要的身分影响机能。例如：基于一个典范的桌面驱动寻址时候要花费9ms，而办事器则仅仅3ms。所以，你必须确保你的磁盘也将是高设备的。

Network performance

对于持久和非持久broker，收集流量速度也将是一个限制机能身分。对于大消息可以紧缩它，最好是建议应用异步发送体式格式。

Hardware specification

忽视

Memory available to the JVM

增长内存JVM的内存大小。应用属性-Xmx。例如：增长JVM内存到512 MB， （-Xmx512MB）

Co-locating the Broker

Overview

因为ActiveMQ供给了自力和内嵌broker的体式格式，若是是应用自力broker体式格式，则至少须要有2个链接端，即producer-broker和broker-consumer。反之，内嵌broker是仅仅可能至少有一个连接端（临盆者或花费者可以作为一个broker内部的原件），如许削减了收集的负载。

下图，应用broker作为临盆者直接发送数据到consumer端，前提是临盆者应用vm://和谈来链接的。

Figure 1.1图显示了producer作为数据的发送端，它经由过程broker发送多量量的消息。在这种场景下，用于broker作为数据的发送端是斗劲完美的。因为消息将会直接的发送到consumer上。而不须要经由过程一个中介。这是一种最简化的体式格式用于VM://传输。

Figure 1.1. BrokerCo-located with Producer

 

vm:// brokerName

你将用一个producer或者consumer经由过程VM://和谈和tcp://和谈链接分别连接到broker上。应用vm://和谈和TCP://和谈是不合的。然而，tcp://用于链接的一个长途的broker端.vm://和谈实际上用于本地经由过程API办法调用（内部的消息通道）链接到内嵌的broker上不须要经由过程收集。同样内嵌的broker运行在同一个JVM中。

例如：一个内嵌的broker的实例经由过程如下创建：具体的请参考vm和谈解析

vm://brokerName

此中brokerName默示一个broker的独一的名称标识。此中以上URL默示创建了一个简单的默认的设备的内嵌broker，若是你想定义正确的设备，然而，最好的体式格式是你须要设置一个brokerConfig选项。

例如：创建一个名称为myBroker的broker的实例来自于设备文件activemq.xml。如下：

vm://myBroker？brokerConfig=xbean:activemq.xml

更多的请参考vm和谈解析

A simple optimization

默认，嵌入式broker的操纵是asynchronous模式，（换句话说，消息是被多线程的分发给consumer）若是你想封闭asynchronous模式，可以削减高低文之间的切换，例如，封闭基于vm的异步发送模式；如下：

vm://brokerName？async=false

Note 若是封闭broker端的optimizedDispatch和consumer端的dispatchAsync的选项的异步行动，则会导致线程将直接分发给consumer。

Optimizing the Protocols

TCP transport

一般的，经由过程增长缓冲大小来提拔TCP和谈的机能。如下：

·       Socket buffer size—默认TCP socket的缓冲大小是 64 KB. 以下法例是评估最佳化的TCP socket的缓冲大小。

Buffer Size = Bandwidth（宽带） x Round-Trip-Time（往返时候）

Round-Trip-Time：发送TCP数据包和接管回执之间花费的时候。

起原参考：http://en.wikipedia.org/wiki/Network_Improvement

例如：

tcp://hostA:61617？socketBufferSize=131072  

·       I/O buffer size—此中I/O 缓冲是用户在TCP和和谈层之上的数据缓冲，默认I/O缓冲大小是8kb，若是你想进步机能，则可以增长该I/O缓冲的大小。

例如：

tcp://hostA:61617？ioBufferSize=16384  

OpenWire protocol

OpenWire主如果用于改变机能的几个参数设置如下：

Table 1.1. OpenWireParameters Affecting Performance

Parameter	Default	Description
cacheEnabled	true	是否缓冲常用的 数据，有利于消息分列编组的优化。
cacheSize	1024	缓冲大小，增长缓冲大小有利于优化消息分列编组的机能。
tcpNoDelayEnabled	false	若是为true，禁用Nagles算法。 Nagles算法设计，以避免发送渺小的TCP数据包包含只有一个或两个字节的数据;例如，当应用Telnet和谈是TCP。若是您禁用Nagles算法，数据包可以被发送更敏捷，但有一个风险是很是小的数据包的数量会增长。
tightEncodingEnabled	true	当为true时，它将实现了基于字节的更紧凑的编码&＃26684;式，成果会导致消息变小和收集机能提拔，然则如许CUP也许要付出额外的花费，这须要调和考量，若是你觉的CPU不是首要的题目的话，可以启用该选项，反之封闭。

重视：

设置以上属性必须应用 wireFormat.*作为前缀.例如：

tcp://hostA:61617？wireFormat.cacheSize=2048  

Enabling compression

若是你发送的消息斗劲大，以及收集斗劲慢，则你须要紧缩消息，jms的消息体（body）被紧缩，heard不被紧缩。如许会使消息变小和收集机能的提拔。另一方面，它同时也会增长CPU的负载。

示例：

// Java  

...  

// Create the connection.  

ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory（user， password， url）;  

connectionFactory.setUseCompression（true）;  

Connection connection = connectionFactory.createConnection（）;  

connection.start（）;  

也可以用 jms.useCompression 选项在URI上设置。

tcp://hostA:61617？jms.useCompression=true  

Message Encoding

Message body type

JMS定义5中类型的消息：

·       StreamMessage

·       MapMessage

·       TextMessage

·       ObjectMessage

·       BytesMessage

BytesMessage 是最快的，ObjectMessage （序列化对象） 是最慢的。

Encoding recommendation

应用 BytesMessage是最好的机能

Consumer Performance

Acknowledgement

确认花费模式的选择对机能有很大的影响，因为每一个确认消息即被经由过程收集发送消息的开销。要进步这面的机能发挥解析的关键，是发送分批确认消息而不是零丁的发送每个消息的确认。

Supportedacknowledgment modes

ActiveMQ 支撑以下消息确认模式:

Session.AUTO_ACKNOWLEDGE

默认的消息确认模式， session 将按照MessageConsumer.receive（） 返回成功或者用于办法回调MessageListener.onMessage（） 返回成功来主动回执消息确认。

Session.CLIENT_ACKNOWLEDGE

在这种模式下，客户端应用法度代码中显式调用Message.acknowledge（）办法，以确认该消息。

Session.DUPS_OK_ACKNOWLEDGE

该模式，jms session可以或许主动的回执消息确认，然则它是一种怠惰的消息确认模式，一旦应用体系故障，一些消息可能会被处理惩罚然则没有回执，一旦体系重启，则会呈现消息的重发题目。

这是一种最快的消息确认模式，然则，花费者必须处理惩罚反复的消息处理惩罚题目。（例如：如何侦测和丢弃重发的消息的处理惩罚逻辑）

Session.SESSION_TRANSACTED

 

当应用事务时，会话会隐含SESSION_TRANSACTED模式。事务提交的反响，然后相当于消息的确认。

当应用JMS事务来发送一组消息（多个消息），交易模式是很是有效的。但避免应用事务发送一个消息，因为这会带来额外的开销提交或回滚事务。

ActiveMQSession.INDIVIDUAL_ACKNOWLEDGE

它不是一个标准的消息确认模式，很类&＃20284;与CLIENT_ACKNOWLEDGE，只可惜它的消息确认是一种办法回调。在完成消息处理惩罚后不会去向理惩罚消息确认的。

optimizeAcknowledge 参数

该参数的功能和DUPS_OK_ACKNOWLEDGE 模式根蒂根基一样；

设置体式格式

1:tcp://hostA:61617？jms.optimizeAcknowledge=true  

或者  

2:connectionFactory.setOptimizeAcknowledge=true  

Optimizing theacknowledgment mode

一般景象下，你要完成最好的机能请求有两种体式格式：

一种你用JMS transactions，基于一个transactions中的批量消息处理惩罚；（即一次履行多个消息）

一种应用 DUPS_OK_ACKNOWLEDGE 模式， 然则该模式须要你去实现如何去除反复的消息的逻辑。可以参考（Apache Camel）http://camel.apache.org/idempotent-consumer.html用于去除反复的消息。

Reducing ContextSwitching

Overview

这里有两种优化线程模式的体式格式：

·       基于broker端的优化消息分发

·       基于consumer端的优化消息接管

Optimize messagedispatching on the broker side

在broker端，默认broker分发消息给consumer是异步的体式格式。也是最好的机能。若是你确信你的consumer老是很快，然而，你将异步改变成同步将会有很好的机能。（削减了高低文切换的花费）

Broker端的asynchronous分发策略的封闭或启用首要根据于花费者。是以，若是你的consumer斗劲快，则封闭asynchronous=false，反之对于慢花费者则asynchronous=true

Consumer端的设置：

TEST.QUEUE？consumer.dispatchAsync=false  

也可以在connectionFactory上设置

// Java  

（（ActiveMQConnectionFactory）connectionFactory）.setDispatchAsync（false）;  

Optimize messagereception on the consumer side

在consumer端，有2个层的线程处理惩罚接管的消息，一个是 Session线程；一个是MessageConsumer线程，在规范中，一个session仅仅接洽关系一个connection。基于某些场景，两个线程同时存在可能是多余的，接下来将评论辩论如何优化它。

Default consumerthreading model

如下图所示，赐与了一个默认的花费者线程接管模式。第一个线程层负责直接把消息从transport层拿过来，进行编组和插入到javax.jms.Session内部的queue中。第二个线程层包含了一个线程池，每个线程接洽关系了一个javax.jms.MessageConsumer 的实例，该层的线程负责从session的queue中拿出消息放入到javax.jms.MessageConsumer 的queue中。

Figure 1.2. DefaultConsumer Threading Model

 

Optimized consumerthreading model

下图，给出了一个最佳化的花费者接管消息线程模式。若是这里仅仅有一个session接洽关系到connection，这种场景下，将直接从transport层把消息直接发送到MessageConsumer 线程上。

Figure 1.3. OptimizedConsumer Threading Model

 

Prerequisites

必须满足以下前提才干起到优化感化：

1.    必须connection上仅仅有一个session相接洽关系。若是存在多个session实例，则alwaysSessionAsync参数设置将无效。

2.    必须应用以下此中一种消息确认模式：

o  Session.DUPS_OK_ACKNOWLEDGE

o  Session.AUTO_ACKNOWLEDGE

alwaysSessionAsyncoption

设置consumer线程优化，必须在ActiveMQConnectionFactory 设置参数alwaysSessionAsync=false ，默认在ActiveMQConnectionFactory上是true

Example

// Java  

...  

// Create the connection.  

ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory（user， password， url）;  

connectionFactory.setAlwaysSessionAsync（false）;  

Connection connection = connectionFactory.createConnection（）;  

connection.start（）;  

// Create the one-and-only session on this connection.  

Session session = connection.createSession（false， Session.AUTO_ACKNOWLEDGE）;  

Prefetch Limit

Overview

如下图，broker将守候接管consumer的消息确认。

Figure 1.4. ConsumerPrefetch Limit

若是花费者的消息确认很慢，broker仍然给它发送消息，会有多量的未确认的消息聚积起来，在这种景象下，broker将不克不及持续给consumer发送消息了。因为多量的未确认的消息已经达到限制了-prefetchlimit。直到该花费者返回消息确认。

Default prefetch limits

Queue consumer

Default prefetch limit is 1000.

若是你用了多个花费者去花费同一个Queue上的消息，此时默认（1000）的大小可能相对斗劲小，须要设置较大些，若是此中一个花费者积攒了多量的未确认的消息，则会使其它花费者接管不到消息。若是花费者失败，多量的未确认的消息必须守候consumer从头恢复后才干处理惩罚。

Queue browser

Default prefetch limit is 500.

Topic consumer

Default prefetch limit is 32766.

这个默认&＃20540;是最大的short类型的数&＃20540;和最公道的数&＃20540;。

Durable topic subscriber

Default prefetch limit is 100.

若是提拔机能话，可以增长该数&＃20540;。

Optimizing prefetch limits

典范的，一个幻想且优化的Queue花费者和持久主题订阅如下所述：

·       Queue consumers—若是你恰好有一个零丁的consumer来花费一个Queue，你须要设置一个相对较大且公道的数&＃20540;。若是你用了多个花费者来花费，此时须要你限制每个花费者的prefetch limit大小，例如：0，1等，包管每个花费者都能轮询到消息。

·       Durable topic subscribers—主题订阅的机能影响主如果prefeth limit的大小，可以试图增大限制。例如：增大到1000.

How to set prefetch limits

以下三种体式格式设置：

·       Per broker.

·       Per connection factory.

·       Per destination.

Per broker

示例：

<broker ... >  

  ...  

  <destinationPolicy>  

    <policyMap>  

      <policyEntries>  

        <policyEntry queue="queue.>" queuePrefetch=”1”/>  

        <policyEntry topic="topic.>" topicPrefetch=”1000”/>  

      </policyEntries>  

    </policyMap>  

  </destinationPolicy>  

  ...  

</broker>  

属性policyEntry 用于规定的prefetch的limint属性名称：

queuePrefetch

queue的prefecth limit

queueBrowserPrefetch

queue的browser 的prefecth limit

topicPrefetch

topic的prefecth limit

durableTopicPrefetch

durable topic subscriber 的prefetch limit.

Per connection factory

示例：

// Java  

ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory（）;  

Properties props = new Properties（）;  

props.setProperty（"prefetchPolicy.queuePrefetch"， "1000"）;  

props.setProperty（"prefetchPolicy.queueBrowserPrefetch"， "500"）;  

props.setProperty（"prefetchPolicy.durableTopicPrefetch"， "100"）;  

props.setProperty（"prefetchPolicy.topicPrefetch"， "32766"）;  

factory.setProperties（props）;  

Per destination

示例：

// Java  

Queue queue =   

  new ActiveMQQueue（"TEST.QUEUE？consumer.prefetchSize=10"）;  

MessageConsumer consumer = session.createConsumer（queue）;  

 

.....未完成。。。