activemq消息大小配置

时间 2019-11-16

原文原文链接

在网上有人问，如何在activemq.xml里面配置消息队列的大小，这样才保证队列不会溢出！若是采用非持久化消息，那么当大量发送失败时候，首先大量占用内存。形成消息堆积，容易形成内存溢出，因此我的比较倾向于持久化消息的同时配合其余方式的master/slave或者failover机制，尽量保持消息的畅通。当咱们开发的Java的使用应用程序的时候，有的时候须要为java应用指定对应的内存大小。当在测试的时候，有的同窗也许会遇到 ActiveMQ报内存不足的问题，咱们能够修改ActiveMQ使用的内存的多少相似Java 的内存的设置，在ActiveMQ.bat或者activemq（linux）文件中html

以下：java

Java代码 mysql

if "%ACTIVEMQ_BASE%" == "" set ACTIVEMQ_BASE=%ACTIVEMQ_HOME%  
  
if "%ACTIVEMQ_OPTS%" == "" set ACTIVEMQ_OPTS=-Xmx512M -Dorg.apache.activemq.UseDedicatedTaskRunner=true -Djava.util.logging.config.file=logging.properties  
  
if "%SUNJMX%" == "" set SUNJMX=-Dcom.sun.management.jmxremote

其中ACTIVEMQ_OPTS中指定ActiveMQ的使用的内存的大小。linux

下面为ActiveMQ的内存分配信息。activemq.xml中的配置状况：sql

Java代码数据库

  <!--  
    The systemUsage controls the maximum amount of space the broker will   
    use before slowing down producers. For more information, see:  
      
    http://activemq.apache.org/producer-flow-control.html  
       
<systemUsage   sendFailIfNoSpaceAfterTimeout="3000"  sendFailIfNoSpace="true">  
    <systemUsage>  
        <memoryUsage>  
            <memoryUsage limit="20 mb"/>  
        </memoryUsage>  
        <storeUsage>  
            <storeUsage limit="1 gb"/>  
        </storeUsage>  
        <tempUsage>  
            <tempUsage limit="100 mb"/>  
        </tempUsage>  
    </systemUsage>  
</systemUsage>  
>

memoryUsage：表示全部队列对象占用的内存大小为20mb;apache

AbstractPendingMessageCursor计算空间的方法源代码以下：缓存

Java代码 oracle

public boolean hasSpace() {  
    return systemUsage != null ? (systemUsage.getMemoryUsage().getPercentUsage() < memoryUsageHighWaterMark) : true;  
}  
  
public boolean isFull() {  
    return systemUsage != null ? systemUsage.getMemoryUsage().isFull() : false;  
}

If you like, you can disable flow control for specific JMS queues and topics on the broker by setting the producerFlowControl flag to false on the appropriate destination policy in the Broker configuration - e.g.app

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

ActiveMQ消息持久化

在broker中设置属性persistent=”true”(默认是true)，同时发送的消息也应该是persitent类型的。ActiveMQ消息持久化有三种方式：AMQ、KahaDB、JDBC。

一、AMQ

AMQ是一种文件存储形式，它具备写入速度快和容易恢复的特色。消息存储在一个个文件中，文件的默认大小为32兆，若是一条消息的大小超过了32兆，那么这个值必须设置大点。当一个存储文件中的消息已经所有被消费，那么这个文件将被标识为可删除，在下一个清除阶段，这个文件被删除。默认配置以下：

Java

1
2
3

</persistenceAdapter>

AMQ的属性：

属性名称	默认值	描述
directory	activemq-data	消息文件和日志的存储目录
useNIO	true	使用NIO协议存储消息
syncOnWrite	false	同步写到磁盘，这个选项对性能影响很是大
maxFileLength	32mb	一个消息文件的大小
persistentIndex	true	消息索引的持久化，若是为false，那么索引保存在内存中
maxCheckpointMessageAddSize	4kb	一个事务容许的最大消息量
cleanupInterval	30000	清除操做周期，单位ms
indexBinSize	1024	索引文件缓存页面数，缺省为1024，当amq扩充或者缩减存储时，会锁定整个broker，致使必定时间的阻塞，因此这个值应该调整到比较大，可是代码中实现会动态伸缩，调整效果并不理想。
indexKeySize	96	索引key的大小，key是消息ID
indexPageSize	16kb	索引的页大小
directoryArchive	archive	存储被归档的消息文件目录
archiveDataLogs	false	当为true时，归档的消息文件被移到directoryArchive,而不是直接删除

二、KahaDB

KahaDB是基于文件的本地数据库储存形式，虽然没有AMQ的速度快，可是它具备强扩展性，恢复的时间比AMQ短，从5.4版本以后KahaDB作为默认的持久化方式。默认配置以下：

Java

1
2
3

</persistenceAdapter>

KahaDB的属性：

property name	default value	Comments
directory	activemq-data	消息文件和日志的存储目录
indexWriteBatchSize	1000	一批索引的大小，当要更新的索引量到达这个值时，更新到消息文件中
indexCacheSize	10000	内存中，索引的页大小
enableIndexWriteAsync	false	索引是否异步写到消息文件中
journalMaxFileLength	32mb	一个消息文件的大小
enableJournalDiskSyncs	true	是否讲非事务的消息同步写入到磁盘
cleanupInterval	30000	清除操做周期，单位ms
checkpointInterval	5000	索引写入到消息文件的周期，单位ms
ignoreMissingJournalfiles	false	忽略丢失的消息文件，false，当丢失了消息文件，启动异常
checkForCorruptJournalFiles	false	检查消息文件是否损坏，true，检查发现损坏会尝试修复
checksumJournalFiles	false	产生一个checksum，以便可以检测journal文件是否损坏。
5.4版本以后有效的属性:
archiveDataLogs	false	当为true时，归档的消息文件被移到directoryArchive,而不是直接删除
directoryArchive	null	存储被归档的消息文件目录
databaseLockedWaitDelay	10000	在使用负载时，等待得到文件锁的延迟时间，单位ms
maxAsyncJobs	10000	同个生产者产生等待写入的异步消息最大量
concurrentStoreAndDispatchTopics	false	当写入消息的时候，是否转发主题消息
concurrentStoreAndDispatchQueues	true	当写入消息的时候，是否转发队列消息
5.6版本以后有效的属性:
archiveCorruptedIndex	false	是否归档错误的索引

从5.6版本以后，有可能发布经过多个kahadb持久适配器来实现分布式目标队列存储。何时用呢？若是有一个快速的生产者和消费者，当某一个时刻生产者发生了不规范的消费，那么有可能产生一条消息被存储在两个消息文件中，同时，有些目标队列是危险的而且要求访问磁盘。在这种状况下，你应该用通配符来使用mKahaDB。若是目标队列是分布的，事务是能够跨越多个消息文件的。

每一个KahaDB的实例均可以配置单独的适配器，若是没有目标队列提交给filteredKahaDB，那么意味着对全部的队列有效。若是一个队列没有对应的适配器，那么将会抛出一个异常。配置以下：

Java

</persistenceAdapter>

</filteredKahaDB>

</persistenceAdapter>

</filteredKahaDB>

</filteredPersistenceAdapters>

</mKahaDB>

</persistenceAdapter>

若是filteredKahaDB的perDestination属性设置为true，那么匹配的目标队列将会获得本身对应的KahaDB实例。配置以下：

Java

</persistenceAdapter>

</filteredKahaDB>

</filteredPersistenceAdapters>

</mKahaDB>

</persistenceAdapter>

三、JDBC

配置JDBC适配器：

Java

1
2
3

</persistenceAdapter>

dataSource指定持久化数据库的bean，createTablesOnStartup是否在启动的时候建立数据表，默认值是true，这样每次启动都会去建立数据表了，通常是第一次启动的时候设置为true，以后改为false。

MYSQL持久化bean

Java

<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>

SQL Server持久化bean

Java

<propertyname="serverName"value="SERVERNAME"/>

<propertyname="portNumber"value="PORTNUMBER"/>

<propertyname="databaseName"value="DATABASE

NAME"/>

<propertyname="user"value="USER"/>

<propertyname="password"value="PASSWORD"/>

</bean>

Oracle持久化bean

Java

<propertyname="driverClassName"value="oracle.jdbc.driver.OracleDriver"/>

<propertyname="url"value="jdbc:oracle:thin:@10.53.132.47:1521:activemq"/>

<propertyname="username"value="activemq"/>

<propertyname="password"value="activemq"/>

<propertyname="maxActive"value="200"/>

<propertyname="poolPreparedStatements"value="true"/>

</bean>

DB2持久化bean

Java

<propertyname="driverClassName"value="com.ibm.db2.jcc.DB2Driver"/>

<propertyname="url"value="jdbc:db2://hndb02.bf.ctc.com:50002/activemq"/>

<propertyname="username"value="activemq"/>

<propertyname="password"value="activemq"/>

<propertyname="maxActive"value="200"/>

<propertyname="poolPrepar