BlazeDS+Spring+activeMQ outofmemory

一、BlazeDS内存溢出修改方案：

BlazeDS在大数据量、频繁推送数据或者频繁刷新页面、且服务器session时间超长的情况下，会导致内存泄漏。由于http协议是无状态的，所以客户端不主动通知blazeds，blazeds是不知道客户端已经断开，一直会等到session超时，且客户端每链接一次服务器，都会产生一个FlexClient，FlexClient对象会维护一个消息队列，所以解决这个问题就应该从两个地方入手，客户端和服务器。
1、flex端代码修改(主动关闭链接)
(1)
//监听页面离开事件

var connectManager:MessageConnectManager = MessageConnectManager.getInstance();
connectManager.regist();
connectManager.addEventListener(MessageConnectManager.DISCONNECT,disConnectHandler);

//页面离开  主动关闭链接

protected function disConnectHandler(e:Event):void
{
if(this.consumer && this.consumer.channelSet){
this.consumer.channelSet.disconnectAll();
}
}

2、修改blazeDS配置文件（网络中断或者用户频繁刷新页面导致内存泄漏）

修改配置文件flex-services-config.xml :

<channel-definition id="my-polling-amf" class="mx.messaging.channels.AMFChannel">
            <endpoint url="http://{server.name}:{server.port}/{context.root}/messagebroker/amfpolling" class="flex.messaging.endpoints.AMFEndpoint"/>
            <properties>
                .....(2)
<flex-client-outbound-queue-processor   class="processor.TimeoutOutBoundQueueProcessor">
<properties>
<!--message time out  in millisecond-->
<messageTimeOut>30000</messageTimeOut>
<!--client time out in millisecond-->
<clientTimeOut>12000</clientTimeOut>
</properties>
</flex-client-outbound-queue-processor>
            </properties>
        </channel-definition>
 
 
<channel-definition id="my-longpolling-amf" class="mx.messaging.channels.AMFChannel">
 
<endpoint url="http://{server.name}:{server.port}/{context.root}/messagebroker/amflongpolling" class="flex.messaging.endpoints.AMFEndpoint"/>
 <properties> 
                .....(3)
<flex-client-outbound-queue-processor   class="processor.TimeoutOutBoundQueueProcessor">
<properties>
<!--message time out  in millisecond-->
<messageTimeOut>30000</messageTimeOut>
<!--client time out in millisecond-->
<clientTimeOut>12000</clientTimeOut>
</properties>
</flex-client-outbound-queue-processor>
     </properties>
</channel-definition> 

二、activeMQ内存溢出修改方案：
acitveMQ在5.4版本以后，能对慢消费者进行处理，比如blazeds+spring+activemq集成时，如果所有客户端与blazeds断开链接，但activemq还会继续向spring JMSAdapter发生数据，且spring JMSAdapter接受到消息后放消息队列里，且不通知activemq，从而导致内存溢出。
修改activemq配置文件

<broker xmlns="http://activemq.apache.org/schema/core"
brokerName="messagebus" dataDirectory="activemq/data"
destroyApplicationContextOnStop="true" persistent="true">
<destinationPolicy>
<policyMap>
<policyEntries> 
      ........(4)
           <policyEntry topic=">" producerFlowControl="false" memoryLimit="1mb" topicPrefetch="100" >
<slowConsumerStrategy>
    <abortSlowConsumerStrategy checkPeriod="30000" maxSlowDuration="120000" abortConnection="false" />
</slowConsumerStrategy>
</policyEntry>
</policyEntries>
</policyMap>
</destinationPolicy>
.......(5)
<systemUsage>
            <systemUsage>
                <memoryUsage>
                    <memoryUsage limit="20 mb"/>
                </memoryUsage>
                <storeUsage>
                    <storeUsage limit="1 gb"/>
                </storeUsage>
                <tempUsage>
                    <tempUsage limit="100 mb"/>
                </tempUsage>
            </systemUsage>
        </systemUsage>
</broker>

配置解释：
(1):客户端监听beforeunload事件，断开链接。工具类见附件“MessageConnectManager.as ”

(2)(3):服务器端保护措施，附件“TimeoutOutBoundQueueProcessor.java ”
     messageTimeOut:队列中消息超时时间，单位毫秒，超过指定时间，剔除队列；
     clientTimeOut：客户端超时时间，如果超过指定时间，blazeDS没有向客户端flush数据，说明客户端已经断开链接，目前测试blazeds最大flush时间1分钟

(4)：activeMQ慢消费者处理策略
topicPrefetch：消费者收到但没有应答的队列长度，超过这个长度，mq就不会向这个消费者推送数据
slowConsumerStrategy：慢消费者策略（如何衡量一个消费者慢？ 在指定时间内未响应消息长度达到topicPrefetch设置或者达到到指定慢的次数，目前使用前者）
checkPeriod：检查消费者是否慢的频度
maxSlowDuration：慢持续时间
abortConnection：是否中断链接（注意 不能中断）

(5)：activeMQ占用系统资源限制，超过限制，持久化到磁盘。

评论

6 楼 keren021 2013-02-01  

coral0212 写道

希望楼主留下一个联系方式，，其他对这个感兴趣的与我联系，我遇到问题了，扣扣158403055

什么问题？ 说说看是否能帮忙！ 不好意思，最近很少看。

5 楼 keren021 2013-01-28  

sharplog 写道

你好，请问你用的是哪个版本的ActiveMQ? 我用5.6.0，参照你的设置了slowConsumerStrategy，但是发现没有效果，可能是什么原因呢？
我在程序里这样创建的subScriber:
this.session.createSubscriber(topic, selector, false);
但我不能在程序中主动去调用subscriber.close()，想通过ActiveMQ的设置，让它自动把一段时间没有读消息的这些subscriber删除掉。就参照你的文章做了相同的配置。但是，我发现并没有效果。
可能是什么原因呢？

一年多没有整这个了，你使用工具查看了吗？

4 楼 coral0212 2012-12-22  

希望楼主留下一个联系方式，，其他对这个感兴趣的与我联系，我遇到问题了，扣扣158403055

3 楼 sharplog 2012-07-07  

你好，请问你用的是哪个版本的ActiveMQ? 我用5.6.0，参照你的设置了slowConsumerStrategy，但是发现没有效果，可能是什么原因呢？
我在程序里这样创建的subScriber:
this.session.createSubscriber(topic, selector, false);
但我不能在程序中主动去调用subscriber.close()，想通过ActiveMQ的设置，让它自动把一段时间没有读消息的这些subscriber删除掉。就参照你的文章做了相同的配置。但是，我发现并没有效果。
可能是什么原因呢？

2 楼 keren021 2012-01-16  

jklliang 写道

轻问
1.var connectManager:MessageConnectManager = MessageConnectManager.getInstance(); 
2.connectManager.regist(); 
3.connectManager.addEventListener(MessageConnectManager.DISCONNECT,disConnectHandler);
MessageConnectManager 这个注册和添加监听是怎样处理的？
如何才能知道页面停止推送

不好意思，最近没有上iteye； 上次附件中漏加“MessageConnectManager”这个文件了，已经补上。“如何才能知道页面停止推送”这个不明白你的意思，链接断开，应该后台就会停止向前端推送数据。

1 楼 jklliang 2011-11-22  

轻问
1.var connectManager:MessageConnectManager = MessageConnectManager.getInstance(); 
2.connectManager.regist(); 
3.connectManager.addEventListener(MessageConnectManager.DISCONNECT,disConnectHandler);
MessageConnectManager 这个注册和添加监听是怎样处理的？
如何才能知道页面停止推送