第十二章 ActiveMQ性能调优 -

zcf9916

浏览: 56829 次
性别:
来自: 福州

最近访客更多访客>>

heavenSea

promiseloney

MartinDon

zmyuit

博主相关

博客

微博

相册

留言

关于我

文章分类

社区版块

存档分类

第十二章 ActiveMQ性能调优

博客分类：

activemq

第十二章 ActiveMQ性能调优

章节导读

学习调优技术
如何优化生产者和消费者
一个优化的实例

ActiveMQ的性能高度依赖于许多不同的因素,包括代理网络的拓扑结构,传输连接器,网络的速度和服务质量,硬件,操作系统以及Java虚拟机.

但是不管在什么应用环境下,你都可以使用一些性能调优技术以改善ActiveMQ性能.也许你的程序不需要确保消息能够准确投送,这时使用可靠的,非持久化消息会有更好的性能.对于减少消息在传输路劲中的序列化次数来说,使用嵌入式代理是很合理的.另外,可以使用大量的调优参数,它们有各自的优点和使用注意事项.

12.1 ActiveMQ通用调优技术

有两种简单方式可以提升JMS消息性能:使用非持久化消息,或者在需要确保消息发送成功时使用事务来将消息分批组合.通常不考虑使用非持久化消息分发除非你不在乎消息可能会丢失(比如,一个实时的数据源,因为状态数据在很短的时间内会重复发送),并且使用事务将消息分批也不总是可行的.但是ActiveMQ为非持久化的消息分发采用了失效安全策略,因此只有灾难性的失败才会导致消息丢失.本节中我们将解释为什么非持久化消息和消息分批会更快,以及为什么在不需要绝对保证消息不会丢失的情况下,可以将它们可应用在你的程序中.

12.1.1 持久化消息 VS 非持久化消息

JMS规范允许两种消息分发模式:持久化分发和非持久化分发.默认的分发模式是持久化分发.当消息生产者发送具有持久化标识的消息到代理时,消息代理将持久化消息到消息存储中然后在发送消息给消费者.这样是为了应对在灾难性失败或者在稍后时间再发送消息给目前尚未激活的消息消费者.

如果你正在使用非持久化消息分发,则JMS允许消息服务提供者使用最大的效率发送消息给当前活动的消息消费者.ActiveMQ对于非持久化消息分发还提供了额外的可靠性保证,本节稍后就此展开介绍.非持久化消息明显比持久化消息快,主要因为下面两点:
(1) 消息生产者通过异步方式发送消息,所以生产者无需等待代理的回执.如图所示.

(2) 持久化消息保存到消息存储(通常需要写磁盘数据)比在网络中传输消息要慢.

使用持久化消息的一个主要理由是在系统崩溃时防止消息丢失.但是正如11章中所述,可以配置ActiveMQ代理的失效转移协议来缓存异步消息,并且在传输连接器失效时重新发送消息(使用失效转移协议的trackMessages属性).同样,ActiveMQ还可以通过客户端或代理端的消息审计以防止消息重复发送.因此,对于各种仅需保证可靠性的应用场景(相对于确保消息能成功送达的场景)来说使用非持久化消息分发模式即可满足你的需求.

因为默认情况下消息分发模式是持久化的,你需要在MessageProducer中明确制定以非持久化方式发送消息,如下代码所示:

MessageProducer producer = session.createProducer(topic);

producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);

我们已经看到为何使用用持久化消息分发和非持久化消息分发会有那么大的性能差异,并且了解了ActiveMQ为改善非持久化消息分发的可靠性而采取的措施.保证消息分发可靠性的好处是允许非持久化消息不仅可用于典型的JMS提供者,而且可以用于更多其他场景中.

12.1.2 事务

当发送消息使用了事务时,只有在事务的边界处(即,Session.commit()方法)会导致与消息代理的同步通信.因此,可以通过批量化的消息生产和消息消费来改善发送持久化消息的性能.下面是相关的代码示例:

public void sendTransacted() throws JMSException {
		ActiveMQConnectionFactory cf = new ActiveMQConnectionFactory();
		Connection connection = cf.createConnection();
		connection.start();
		Session session = connection.createSession(true, Session.SESSION_TRANSACTED);
		Topic topic = session.createTopic("Test.Transactions");
		MessageProducer producer = session.createProducer(topic);
		int count = 0;
		for (int i = 0; i < 1000; i++) {
			Message message = session.createTextMessage("message " + i);
			producer.send(message);
			if (i != 0 && i % 10 == 0) {
				session.commit();
			}
		}

	}

	public void sendNonTransacted() throws JMSException {
		ActiveMQConnectionFactory cf = new ActiveMQConnectionFactory();
		Connection connection = cf.createConnection();
		connection.start();
		// create a default session (no transactions)
		Session session = connection.createSession(false, Session.AUTO_ACKNOWELDGE);
		Topic topic = session.createTopic("Test.Transactions");
		MessageProducer producer = session.createProducer(topic);
		int count = 0;
		for (int i = 0; i < 1000; i++) {
			Message message = session.createTextMessage("message " + i);
			producer.send(message);
		}

	}

至此我们已经了解了性能调优的几个简单方式:如果可行的化使用非持久化消息,以及上面介绍的,在合适的情况下,在应用程序中使用事务边界发送持久化消息.

12.1.3 嵌入式代理

通常有的需求要求应用程序和代理在同一位置(译注:虚拟机)部署,因此所有依赖消息代理的服务只有在消息代理可用的时候才可用,如图所示.创建一个嵌入式代理相当容易,并且使用虚拟机传输连接器的一个优势是通过嵌入式代理分发的消息不会因网络传输要求的序列化而产生性能损耗,这就使得嵌入式代理成为那些要求能对大量服务做出快速响应的程序的理想选择.

你可以使用一个监听TCP连接的传输连接器来创建一个嵌入式代理,并且依然可以使用VM传输连接器来连接到这个代理.默认情况下,所有代理都监听使用vm://<broker name>的传输连接.下面的示例代码是一个示例,该示例中设置了一个使用嵌入式代理监听一个名称为service.queue的队列请求的服务.

	        BrokerService broker = new BrokerService();
		broker.setBrokerName("service");
		broker.setPersistent(false);
		broker.addConnector("tcp://localhost:61616");
		broker.start();
		ActiveMQConnectionFactory cf = new ActiveMQConnectionFactory("vm://service");
		cf.setCopyMessageOnSend(false);
		Connection connection = cf.createConnection();
		connection.start();
		Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
		final MessageProducer producer = session.createProducer(null);
		Queue queue = session.createQueue("service.queue");
		MessageConsumer consumer = session.createConsumer(queue);
		consumer.setMessageListener(new MessageListener() {
			public void onMessage(Message msg) {
				try {
					TextMessage textMsg = (TextMessage) msg;
					String payload = "REPLY: " + textMsg.getText();
					Destination replyTo = msg.getJMSReplyTo();
					textMsg.clearBody();
					textMsg.setText(payload);
					producer.send(replyTo, textMsg);
				} catch (JMSException e) {
					e.printStackTrace();
				}
			}

		});

你可以使用一个javax.jms.QueueRequestor使用TCP传输连接器来连接到上述嵌入式代理提供的服务,如下代码所示:

	ActiveMQConnectionFactory cf = new ActiveMQConnectionFactory("tcp://localhost:61616");
		QueueConnection connection = cf.createQueueConnection();
		connection.start();
		QueueSession session = connection.createQueueSession(false, Session.AUTO_ACKNOWLEDGE);
		Queue queue = session.createQueue("service.queue");
		QueueRequestor requestor = new QueueRequestor(session, queue);
		for (int i = 0; i < 10; i++) {
			TextMessage msg = session.createTextMessage("test msg: " + i);
			TextMessage result = (TextMessage) requestor.request(msg);
			System.err.println("Result = " + result.getText());

		}

默认情况下,ActiveMQ总是复制消息发送者发送的真实消息以隔离消息在通过代理传输时可能发生的改变以及消息被代理处理后发生的改变,消息发送和消息拷贝在同一个Java虚拟机中完成.如果你不打算使用保留发送时状态的原始消息,你可以通过设置ActiveMQConnectionFactory的copyMessageOnSend属性为false来避免消息发送时复制消息的开销,示例代码如下所示:

ActiveMQConnectionFactory cf = new ActiveMQConnectionFactory();

cf.setCopyMessageOnSend(false);

我们已经了解了ActiveMQ的一些比较容易实现的提升性能的技术.使用一个和应用程序部署在一起的嵌入式代理是一种很微小的结构变化.获取的性能提升和与代理一起部署的服务的原子性也是这种架构变化有吸引力的原因.

12.1.4 优化OpenWire协议

OpenWire协议的一些选项参数值得关注.OpenWire协议是一些二进制格式规范,传输连接器(例如)使用这种格式的传输命令传输数据到代理.这些命令包含消息相关命令和消息确认相关命令,以及代理相关的管理和控制命令.表13.1展示了一些OpenWire协议与性能调整有关的包装格式参数.

属性名称默认值描述

tcpNoDelayEnabled	false	通知传输连接器的对端是否启用 tcpNoDelay(tcp 无延迟).如果启用,在你通过相对较慢的网络发送大量小数据类型消息时提升性能.
cacheEnabled	true	如果启用,则缓存重复的值(比如producerId和消息目的地),允许传输与缓存值对应的简短的key.该设置减小了传输数据的大小,因而在网络性能不佳时可以提升性能.但是因为在缓存中查找数据会同时在客户端和代理所在的机器中引入CPU负载的额外开销,配置时请考虑中引入开销的影响.
cacheSize	1024	缓存条目的最大数量.该值不能超过Short.MAX_VALUE值的二分之一.当开启缓存时,该值设置的越大性能越好.但是因为每一个传输连接都独立使用一个缓存,所以需要在代理端考虑因缓存带来的额外开销,特别是当有大量的客户端连接到代理时.
tightEncodingEnabled	true	以CPU敏感的方式压缩消息.我们推荐在代理打算使用所有可用CPU时将该选项设置为停用.

你可以通过下面的方式,将上述参数附加到连接到代理的传输连接器的URI中.下面的示例代码中展示了如何使用tightEncodingEnabled参数禁用紧凑编码(tight encoding).

String uri =

"failover://(tcp://localhost:61616?wireFormat.cacheEnabled=false&\

wireFormat.tightEncodingEnabled=false)";

ActiveMQConnectionFactory cf = new ActiveMQConnectionFactory(url);

cf.setAlwaysSyncSend(true);

12.1.5 优化TCP传输连接器

ActiveMQ中使用最广泛的连接器是TCP传输连接器.有两个直接影响TCP传输连接器的性能,它们是:
(1)socketBufferSize–TCP传输连接器用于发送和接收数据的缓冲区大小.通常该参数设置的越大越好(尽管这个最大值收到操作系统限制,但是可以去测试).默认值为65536,单位是byte.
(2)tcpNoDelay–默认值为false.通常TCP套接字缓存即将被发送的小尺寸数据包.当启用这个参数时,消息会被尽快发送(译注:不缓冲).同样可以测试这个配置,因为修改这个参数是否能提升性能还和操作系统有关.

下面是一个示例代码,演示了在传输连接器的URI中启用tcpNoDelay参数:

String url = "failover://(tcp://localhost:61616?tcpNoDelay=true)";
ActiveMQConnectionFactory cf = new ActiveMQConnectionFactory(url); 
cf.setAlwaysSyncSend(true)

12.2 优化消息生产者

在代理分发消息到消息消费者之前,消息生产者发送消息到代理的效率是影响整个应用程序性能的基本元素.下面我们将关注几个消息生产者发送消息给代理时,影响消息吞吐量和消息发送延迟的调优参数.

12.2.1 异步发送

我们已经了解了使用非持久化消息分发可以带来性能提升.ActiveMQ中非持久化消息分发是可靠的,因为消息会在网络故障和系统崩溃(只要消息生产者依然是活动的–此时消息生产者将消息缓存在失效转移连接器的缓存中)中幸存下来.但是,通过设置消息生产者的连接工厂的useAsyncSend属性,你仍然可以在使用持久化消息时获得性能提升,请看下面示例代码:

ActiveMQConnectionFactory cf = new ActiveMQConnectionFactory(); 
    cf.setUseAsyncSend(true);

上面代码将设置一个属性,告知消息生产者不要尝试从代理获取一个刚刚发送的消息的回执. 这就是说消息生产者在等待消息持久化到磁盘之后就不再等待(译注:消息回执)了而是继续发送下一条消息.

如果你的程序要求消息精确分发,则推荐你使用系统默认的持久化消息分发模式,并且最好同时使用事务.

使用异步方式发送消息获以取性能提升的原因不难理解,同时以这种方式实现性能提升的方式也很简单–只要设置ActiveMQ连接工厂的一个属性即可.下一节将介绍ActiveMQ中一种通常不被理解的特性:消息生产者流控制.我们将看到大量的关于消息生产者效率下降或暂停问题,并且理解如何在你的应用程序中使用流控制来减少这些问题的发生.

12.2.2 生产者流控制

生产者流控制允许消息代理在系统资源紧张时降低消息的通过量.这种情况在消息消费者处理速度慢于生产者时发生,此时消息会缓存在代理的内存中等待被发送.

消息生产者在收到代理通知有足够存储空间接收更多消息之前都会处于等待状态,如图13.3所示.生产者流控制对于阻止代理的内存和临时存储空间超过限制的值来说是必要的,尤其是对于广域网来说.

对于持久化消息来说,生产者流控制默认时就是开启的,但是对于异步消息发布来说必须明确的指定为开启(针对持久化消息,或对于配置成总是异步发送消息的连接来说).你可以通过设置连接工厂的producerWindowSize属性来开启消息生产者异步发送消息的流控制.

ActiveMQConnectionFactory cf = new ActiveMQConnectionFactory();
cf.setProducerWindowSize(1024000);

producerWindowSize属性用于设置生产者在收到代理的存储空间未超过限制值回执之前可以用来缓存消息的字节大小,超过这个设置的值,生产者将等待代理回执(译注:而不再生产和发送消息).对于一个异步发送消息的生产者来说,如果这个属性未开启,则代理仍然会暂停消息流动,这种情况下,默认会阻塞消息生产者的传输连接器.阻塞传输连接器会阻塞所有使用该连接器连接的用户,如果消息消费者也使用同样的连接,则会导致死锁.生产者流控制运行仅阻塞消息生产者而不是整个消息生产者使用的传输连接.

尽管保护代理不要运行在可用内存空间低的状态是一个不错的想法,但是这并不能改善系统被最慢的消费者而拖慢时而产生的性能问题.所以,让我们看看禁用生产者流控制时会发生什么,如下面代码中的粗体字所示.你可以在代理的配置中配置消息目的地策略来实现.

    <destinationPolicy>
	<policyMap>
		<policyEntries>
			<policyEntry topic="FOO.>"  producerFlowControl="false" memoryLimit="10mb" />
			</policyEntries>
		</policyMap>
	</destinationPolicy>

生产者流控制调优

默认情况下,禁用了生产者流控制之后,发送给速度慢的消费者的消息会被存放到临时存储控件中,以便消息生产者和其他消息消费者可以尽可能快的运行,如图所示.另外,代理可用的最大内存限制决定了在什么时候消息会通过追加到消息游标的形式持久化到磁盘上.系统可用最大内存限制作用范围是整个代理.这个限制应该要低于消息目的地可用内存限制,以便在流控制之前起作用.(译注:即,这个较小的值先起作用,则消息目的地使用内存不会超过配置的限制值,因为这个值较大)

尽管有一些因为存储消息而代理的性能损失,在禁用了生产者流控制之后,消息应用程序可以独立于最慢的消息消费者而运行.在理想情况下,消息消费者总是与消息生产者一样快速运行,这就给我们引入下一节中关于消息消费者的优化.

默认情况下,当启用了生产者流控制后,当代理没有空间存放更多消息时,生产者发送消息的操作会被阻塞直到代理有足够控件存储消息.有两中方式调整该参数,使得代理获取更多存储消息控件之前,消息生产者不会无限期实质性的挂起.
第一种调节消息生产者流控制的方式称为sendFailIfNoSpace

<systemUsage>
	<systemUsage sendFailIfNoSpace="true">
		<memoryUsage>
			<memoryUsage limit="128 mb"/>
		</memoryUsage>
	</systemUsage>
</systemUsage>

sendFailIfNoSpace属性将控制权返还给消息生产者,在代理的消息存储已经不足而生产者仍然尝试发送操作时,通过在生产者客户端抛出异常来代替永久性的阻塞生产者的发送操作.这就允许生产者可以捕捉这个异常,然后等待一段时间后,继续尝试发送操作.

第二个调节生产者流控制的属性开始于ActiveMQ的5.4.1版本.该属性名称为sendFailIfNoSpaceAfterTimeout:

<systemUsage>
	<systemUsage sendFailIfNoSpaceAfterTimeout="5000">
		<memoryUsage>
			<memoryUsage limit="128 mb"/>
		</memoryUsage>
	</systemUsage>
</systemUsage>

sendFailIfNoSpaceAfterTimeout与前面那个属性稍有不同.配置了该属性后,在等待配置的时间后,如果代理端依然没有足够的空间存储消息,则会在客户端客户端发送消息时抛出异常.

12.3 优化消息消费者

为了最大限度的提升应用程序的性能,你必须关注所有影响性能的因素.到目前为止,消息消费者在整个ActiveMQ系统的性能表现中都扮演着举足轻重的角色.通常,消息消费者必须要尽量以2倍于消息生产者的速度运行,因为消费者还要通知代理消息已经被处理了.下面我们将介绍通过优化消息消费者你可以获取的最大的性能提升.

通常,ActiveMQ代理会通过消费者连接尽可能快的发送消息.通常情况下,一旦消息通过ActiveMQ代理的传输连接发送完成之后,消息就加入了与消费者关联的session队列中,并等待分发.在下一节中,我们将解释消息发送给消费者的速度为何可控以及如何控制,同时还将阐述如何调整这个消息发送速率已获取更好的吞吐量.

12.3.1 预获取限制

ActiveMQ使用一种基于推送的模式来将收到的消息分发给代理.为了防止超过消费者的内存限制,有一个参数(prefetchlimit)可以限制代理在消费者确认消息已被应用程序处理之前可以发送给消费者的消息数量.在消费者内部,从传输连接器上接管的消息会被分发并放置于一个和消费者 session管理的内部队列中

消费者连接会在内部将分发过来的消息队列化.这个内部的消息队列的尺寸加上尚未发送回执给代理的消息(这些消息已经被消费者接收了但是还没有通知代理消息已被消费)的数量之和受到消费者的prefetchlimit参数限制.通常,这个prefetchlimit参数设置的越大,消费者运行的越快.

但是对于消息队列来说,设置这个限制并非是最理想方案,因为你可能希望消息会被平均的分发给一个队列上的所有消费者.这种情况下,当prefetchlimit设置的很大时,处理速度较慢的消费者可能会累积待处理的消息,而这些消息却不能被更快的消费者处理.这种情况下,设置较低的prefetchlimit值可能会更适合.如果prefetchlimit值设置为0,这消息消费者会主动从代理拉取消息并且不允许代理推送任何消息到消费者.

对于不同种类的消费者而言有不同的prefetch limit默认值:
队列消费者的prefetch limit默认值为1000
队列浏览消费者的prefetch limit默认值为500
持久化主题消费者的prefetch limit默认值为100
非持久化主题的prefetch limit默认值为32766

prefetch limit值是消息消费者等待接收的消息的数量而不是内存值大小.可以通过设置 ActiveMQConnectionFactory的相关属性值值来设置prefetch limit,如下代码所示:

代码清单13.11 配置ActiveMQConnectionFactory的预拉取策略

ActiveMQConnectionFactory cf = 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"); 
cf.setProperties(props);

或者,在创建一个消息目的地时,传递预拉取尺寸参数作为消息目的地的属性,如下代码所示:创建消息目的地时设置预拉取尺寸

Queue queue = new ActiveMQQueue("TEST.QUEUE?consumer.prefetchSize=10"); 
    MessageConsumer consumer = session.createConsumer(queue);

使用Prefetchlimit是一种简单的提升性能机制,但是需要谨慎使用.对于队列来说,你应该考虑你的程序中是否有比较慢的消费者;而对于主题来说,你需要考虑在消息被分发之前,你的队列可以使用的客户端上的最大内存是多少.

控制消息分发给消费者的速率仅仅是消费者调优的一部分.一旦消息到达消费者的连接器之后,消息分发到消费者时使用的方法以及消费者用来将消息已被处理的确认发送给代理时使用的选项就成为影响性能的重要组成部分.我们将在下一节讨论相关内容:

12.3.2 消息的分发和确认

由图13.5可以看出,使用javax.jms.MessageListener.onMessage()来分发消息明显比使用 javax.jms.MessageConsumer.receive()要快.如果MessageConsumer没有设置MessageListener则该消费者的消息会分发到队列中然后等待调用receive()方法.不仅维护消费者内部队列的代价是昂贵的,而且应用程序线程不断的调用receive()来切换应用程序上下文的代价也是高昂的.

因为ActiveMQ代理需要保存一个记录以表明当前有多少消息已被消费这消费了来维护消费者内部的prefetchlimit,MessageConsumer必须为每一个消息发送消息确认.如果你使用了事务, 当调用Session.commit()方法是会发送消息确认,但是假如你使用auto-acknowledgment模式则每个消息处理完成后都会独自发送消息确认.

有一些优化选项专门用于发送消息确认给代理,当使用DUPS_OK_ACKNOWLEDGE session确认模式时, 这些优化选项可以显著的改善性能.另外,你可以设置ActiveMQ ConnectionFactory的optimizeAcknowledge属性,通过给消费者一个提示信息以便批量发送消息确认信息.
代码清单13.13 设置optimizeAcknowledge属性

ActiveMQConnectionFactory cf = new ActiveMQConnectionFactory(); 
    cf.setOptimizeAcknowledge(true);

当在一个session中使用optimizeAcknowledge或DUPS_OK_ACKNOWLEDGE确认模式时,消费者只发送一个消息告知ActiveMQ代理一批消息已经完成处理.这样消息消费者要做的工作就减少了,便于消费者尽可能快的处理消息.

下面的列出了确认消息的不同选项以及使用这些选项后消费者发挥消息确认给ActiveMQ代理的频率.

ActiveMQ消息确认模式

Session.SESSION_TRANSACTED 使用 Session.commit()方法批量确认这是消息消费的一种可靠方式,并且性能很好,允许消息一次提交中处理多个消息.

Session.CLIENT_ACKNOWLEDGE 当一个消息确认了所有消息都确认在确认之前可以使消费者消费大量的消息

Session.AUTO_ACKNOWLEDGE 每个消息处理完成后自动发送这种方式会比较慢,但常常是默认的消息确认机制消息确认到代理.

Session.DUPS_OK_ACKNOWLEDGE 允许消息消费者发送确认信息给代理有prefetch limit值一半的消息已经被消费当消费者收到的消息达到prefetch limit的一半时即发送消息确认给代理.这消息处理中最快的标准的消息确认方式.

ActiveMQSession.INDIVIDUAL_ACKNOWLEDGE每处理一个消息就发送一次确认最大限度的允许控制每个消息独立的被确认,但是会很慢.

optimizeAcknowledge 允许消息消费者发送确认信息给代理有一定范围内的消息已经被消费与Session.AUTO_ACKNOWLEDGE一同起作用,在消费者处理完的消息占到prefetch缓存的 65%时发送消息确认.使用这种模式可以以最快的方式处理消息.

不单独确认每个消息的缺点是,不管消息消费者以任何理由失去了与ActiveMQ代理连接，那么你的消息应用程序可能会收到重复的消息.但是,对于要求快速处理且不关心消息是否重复的应用程序(比如实时的数据源)来说,推荐使用optimizeAcknowledge模式.

ActiveMQ的消息消费者包含重复消息侦测机制,可以最大限度的降低收到重复消息的风险.

12.3.3 异步分发

每个session都维护一个内部的即将被分发到各自的消费者的消息的队列(如图13.5所示).内部消息队列以及与之关联的用于发送消息到消息消费者的线程的使用可能会给消息处理增加额外开销.

你可以禁用ActiveMQ连接工厂的alwaysSessionAsync属性来停用上述消息队列和消息分发线程.这种设置运行消息直接从传输连接器发送到消息消费者.下面代码是禁用该属性的示例代码:

代码清单13.14 禁用alwaysSessionAsync属性

ActiveMQConnectionFactory cf = new ActiveMQConnectionFactory();
cf.setAlwaysSessionAsync(false);

停用asynchronous允许消息直接发送到session内部的队列并由session负责进一步分发, 如图13.6所示.

到目前为止,我们已经了解了一些通用的提升性能技巧,比如使用可靠的消息系统以替代能确保消息收发系统以及使用和应用程序同址部署的ActiveMQ代理(译注:嵌入式代理).我们也了解了传输连接器,消息生产者和消费者的一些不同的调优参数.

使用示例是解释问题的最好方式,因此下一节中我们将使用一个实时的数据源程序示例来演示如何改善性能.

12.4 ActiveMQ性能优化实战

让我们通过一个例子来看看如何综合使用前面介绍的性能调优方法.我们将用程序模拟一个实时的数据源,该程序中消息生产者和一个嵌入时代理部署在一起,同时使用消息消费者监听远程的消息.

我们将阐如何述使用一个嵌入式代理来减少将消息发送到ActiveMQ代理的开销.我们还将调整消息消费者的一些选项来降低消息的拷贝.嵌入式代理将被配制成禁用流控制并且使用内存限制以允许代理快速处理消息流.

最后,消息消费者将会配置成直接通过分发方式,同时配置一个高prefetch limit值以及配置优化过的消息确认模式.

首先我们设置一个嵌入式代理,设置其可用内存限制为一个合理的值(64M),为每一个消息目的地设置可用内存限制,并且停用消息生产者流控制.

如下代码所示,使用默认的PolicyEntry设置代理的消息目的地策略.PolicyEntry保存了ActiveMQ代理的消息目的地的相关配置信息.你可以为每一个消息目的地单独设置策略,也可以使用通配符将一个策略应用到多个配置通配符的消息目的地(比如,名称为foo.>的PolicyEntry将仅应用到名称以foo开头的消息目的地).在我们的例子中,我们仅仅设置内存限制以及禁用生产者流控制.为了简单起见,我们仅仅配置了默认的策略实体,该策略实体将应用到所有消息目的地.

BrokerService broker = new BrokerService();
broker.setBrokerName("fast");
broker.getSystemUsage().getMemoryUsage().setLimit(64*1024*1024);
PolicyEntry policy = new PolicyEntry();
policy.setMemoryLimit(4 * 1024 *1024);
policy.setProducerFlowControl(false);
PolicyMap pMap = new PolicyMap();
pMap.setDefaultEntry(policy);
broker.setDestinationPolicy(pMap);
broker.addConnector("tcp://localhost:61616");
broker.start();

上面代码创建的代理使用了一个唯一的名称fast,因此与代理同处于一个虚拟机内的数据源生产者可以使用VM传输连接诶其绑定到该代理.

除去使用了嵌入式代理,消息生产者也是简易的,除了将其配置成发送非持久化消息并且不使用消息拷贝.消息生产者的示例代码如下所示:

ActiveMQConnectionFactory cf = new ActiveMQConnectionFactory("vm://fast");
cf.setCopyMessageOnSend(false);
Connection connection = cf.createConnection();
connection.start();
Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
Topic topic = session.createTopic("test.topic");
final MessageProducer producer = session.createProducer(topic);
producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);
for (int i =0; i < 1000000;i++) {
TextMessage message = session.createTextMessage("Test:"+i);
producer.send(message);
}

消息消费者被配置成直通方式(禁用了异步session分发)并使用了javax.jms.MessageListener.消息消费者使用的消息确认模式为optimizeAcknowledge,以便能尽可能快的处理消息.示例代码如下所示:

ActiveMQConnectionFactory cf =  new ActiveMQConnectionFactory("failover://(tcp://localhost:61616)");
cf.setAlwaysSessionAsync(false);
cf.setOptimizeAcknowledge(true);
Connection connection = cf.createConnection();
connection.start();
Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
Topic topic = session.createTopic("test.topic?consumer.prefetchSize=32766");
MessageConsumer consumer = session.createConsumer(topic);
final AtomicInteger count = new AtomicInteger();
consumer.setMessageListener(new MessageListener() {
public void onMessage(Message message) {
TextMessage textMessage = (TextMessage)message;
try {
if (count.incrementAndGet()%10000==0)
System.err.println("Got = " + textMessage.getText());
} catch (JMSException e) {
e.printStackTrace();
}
}
});

本节中,我们使用ActiveMQ综合了各种调优方法创建了一个分布式实时数据源示例程序.我们创建了一个消息生产者demo并将其配置成直接传递消息到一个嵌入式代理.我们还创建了嵌入式代理,同时禁用了其生产者流控制功能.最后我们配置了一个消息消费者尽可能快的接收消息.

我们推荐你尝试修改其中的一些配置参数值(比如optimizeAcknowledge属性值)来看看配置修改对性能有什么影响