在网上有人问,如何在activemq.xml里面配置消息队列的大小,这样才保证队列不会溢出!如果采用非持久化消息,那么当大量发送失败时候,首先大量占用内存。造成消息堆积,容易造成内存溢出,所以个人比较倾向于持久化消息的同时配合其他方式的master/slave或者failover机制,尽量保持消息的畅通。当我们开发的Java的使用应用程序的时候,有的时候需要为java应用指定对应的内存大小。当在测试的时候,有的同学也许会遇到ActiveMQ报内存不足的问题,我们可以修改ActiveMQ使用的内存的多少类似Java 的内存的设置,在ActiveMQ.bat或者activemq(linux)文件中
如下:
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的使用的内存的大小。
下面为ActiveMQ的内存分配信息。activemq.xml中的配置情况:
<!-- 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;
AbstractPendingMessageCursor计算空间的方法源代码如下:
public boolean hasSpace() { return systemUsage != null ? (systemUsage.getMemoryUsage().getPercentUsage() < memoryUsageHighWaterMark) : true; } public boolean isFull() { return systemUsage != null ? systemUsage.getMemoryUsage().isFull() : false; }
ActiveMQ消息持久化
在broker中设置属性persistent=”true”(默认是true),同时发送的消息也应该是persitent类型的。ActiveMQ消息持久化有三种方式:AMQ、KahaDB、JDBC。
1、AMQ
AMQ是一种文件存储形式,它具有写入速度快和容易恢复的特点。消息存储在一个个文件中,文件的默认大小为32兆,如果一条消息的大小超过了32兆,那么这个值必须设置大点。当一个存储文件中的消息已经全部被消费,那么这个文件将被标识为可删除,在下一个清除阶段,这个文件被删除。默认配置如下:
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,而不是直接删除 |
2、KahaDB
KahaDB是基于文件的本地数据库储存形式,虽然没有AMQ的速度快,但是它具有强扩展性,恢复的时间比AMQ短,从5.4版本之后KahaDB做为默认的持久化方式。默认配置如下:
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,那么意味着对所有的队列有效。如果一个队列没有对应的适配器,那么将会抛出一个异常。配置如下:
如果filteredKahaDB的perDestination属性设置为true,那么匹配的目标队列将会得到自己对应的KahaDB实例。配置如下:
1
2
3
4
5
6
7
8
9
10
11
12
|
<persistenceAdapter>
<mKahaDB directory="${activemq.base}/data/kahadb">
<filteredPersistenceAdapters>
<!--kahaDB per destinations-->
<filteredKahaDB perDestination="true">
<persistenceAdapter>
<kahaDB journalMaxFileLength="32mb"/>
</persistenceAdapter>
</filteredKahaDB>
</filteredPersistenceAdapters>
</mKahaDB>
</persistenceAdapter>
|
3、JDBC
配置JDBC适配器:
dataSource指定持久化数据库的bean,createTablesOnStartup是否在启动的时候创建数据表,默认值是true,这样每次启动都会去创建数据表了,一般是第一次启动的时候设置为true,之后改成false。
MYSQL持久化bean
SQL Server持久化bean
Oracle持久化bean
1
2
3
4
5
6
7
8
|
<bean id="oracle-ds"class="org.apache.commons.dbcp.BasicDataSource"destroy-method="close">
<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
相关推荐
ActiveMQ 消息队列
ActiveMQ消息队列主题订阅Spring整合
SpringBoot整合ActiveMQ消息队列和双向队列、点对点与发布订阅,可以参考我的博客文章进行学习https://blog.csdn.net/sujin_/article/details/82956386
可用于调试MSMQ、RabbitMQ、ActiveMQ三种消息队列 其中MSMQ支持Active、Binary、XML格式(要勾选事务) RabbitMQ支持逐条接发、批量接发、RPC回调模式、新建队列、建立持久化队列、连接测试等功能。
SpringBoot快速玩转ActiveMQ消息队列,jdk8下的简要版介绍。
ActiveMQ消息队列的简单实例和注解。附有文档说明。
ActiveMQ队列消息过期时间设置和自动清除解决方案.docx
百度spring整合activemq 发现几乎都只是在xml文件配置固定的消息队列而且太麻烦。并没有根据需求进行动态生成主题和队列。本文档参考了纯粹的activemq java代码和百度上的demo,很简洁的实现了动态消息队列的生成和...
详细描述了ActiveMQ消息过期-时间设置和自动清除解决方案。
ActiveMQ 是Apache出品,最流行的,能力强劲的开源消息总线。ActiveMQ 是一个完全支持JMS1.1和J2EE 1.4规范的 JMS Provider实现,尽管JMS规范出台已经是很久的事情了,但是JMS在当今的J2EE应用中间仍然扮演着特殊的...
使用activeMq开发的一个简单demo,供入门者学习参考用。
该demo在本地启动activemq后放在tomcat中可直接运行,不需要修改配置,连接远程的activemq需要手动修改applicationContext.xml中的activemq连接地址,jdk1.8
1、介绍ActiveMQ5.x消息队列基础特性和本地快速安装 2、SpringBoot2.x整合ActiveMQ实战之点对点消息
这是一个关于activeMQ如何使用的简单案例,使用时可以套用
整合Spring + ActiveMQ 的朋友可以下载看一下 简单易懂
7道消息队列ActiveMQ面试题分享给你! 大家面试前,必须狠刷面试题,哥给你们整理了面试过程中必问的一些面试,希望对你们有帮助,祝你们早日找到满意的工作。
springboot整合activeMq的使用,队列,主题,消息手动确认,重发机制
ActiveMQ5.13 安装与配置
消息队列activemq运用的dome,消息队列的基本实现例子,帮助更快学习消息队列
ActiveMQ的队列模式,初学ActiveMQ的,一个很好的体验、入门,代码注释很清楚,JMS的java使用步骤