消息postgresql9.0

金蝶Apusic消息中间件V9.
0用户手册金蝶天燕中间件股份有限公司目录第一章前言11.
1编写目的.
11.
2术语.
11.
3参考文档.
1第二章安装指南.
22.
1系统要求.
22.
2安装.
22.
2.
1获取安装包.
22.
2.
2安装.
22.
2.
3配置环境变量.
22.
3目录结构.
32.
4卸载.
4第三章快速开始.
53.
1启动/停止服务.
53.
1.
1启动服务.
53.
1.
2停止服务.
53.
2版本信息.
53.
3登陆管理控制台53.
4创建队列.
63.
5发送消息.
7第四章ApusicMQ介绍84.
1ApusicMQ的原理84.
1.
1JMS的基本构件84.
1.
1.
1连接工厂.
84.
1.
1.
2会话84.
1.
1.
3目的地84.
1.
1.
4生产者94.
1.
1.
5消息消费者.
94.
1.
1.
6消息94.
1.
2JMS的可靠性机制.
104.
1.
2.
1消息确认.
104.
1.
2.
2持久性114.
1.
2.
3优先级114.
1.
2.
4消息过期.
114.
1.
2.
5临时目的地.
114.
1.
2.
6持久订阅.
114.
1.
2.
7本地事务.
124.
2ApusicMQ基本配置.
124.
2.
1配置ApusicMQ端口124.
2.
2监控ApusicMQ134.
3ApusicMQ的Broker.
134.
3.
1RunningBroker:144.
3.
2EmbeddedBroker:144.
4ApusicMQBroker的Transport.
154.
4.
1TCPTransport164.
4.
2FailoverTransport.
164.
4.
3DiscoverTransport.
184.
5ApusicMQBroker的持久方式194.
5.
1KahaMessageStore204.
5.
2JDBCMessageStore204.
6ActimveMQ的其他特性214.
6.
1异步发送消息.
214.
6.
2消费者异步分派.
224.
6.
3消费者优先级.
224.
6.
4独占消费者.
234.
6.
5消息重发策略.
234.
6.
6目标相关的属性.
244.
6.
7消息预取处理.
254.
6.
8配置连接URL264.
6.
9消息重发与死信管理.
274.
6.
10生产者和消费者的密码使用29第五章ApusicMQ集群方式.
305.
1AMQ集群特点.
305.
2AMQ集群方式--Queueconsumerclusters.
305.
3AMQ集群方式--Brokerclusters.
315.
3.
1管道订阅conduitSubscriptions335.
3.
2双向网络连接(duplexnetworkConnector)335.
3.
3指定和限制Destination345.
3.
4被卡住的消息.
355.
3.
5其他说明.
365.
4AMQ集群方式--MasterSlave.
365.
4.
1PureMasterSlave.
375.
4.
2SharedFileSystemMasterSlave375.
4.
3JDBCMasterSlave37第六章ApusicMQ的JMX监控.
396.
1JMX的配置396.
2JMX的JAVA端获取信息.
39金蝶Apusic消息中间件v9.
0用户手册第一章前言1.
1编写目的金蝶Apusic消息中间件(ApusicMessageQueue),简称AMQ或ApusicMQ.
该文档主要针对用到AMQ的管理人员,开发人员,还有对AMQ产品进行评估的人员.
它的主要内容是:消息中间件概念描述,AMQ产品介绍,AMQ的管理和配置.
该文档主要侧重于特性介绍和管理功能介绍.
1.
2术语表1:术语词汇与本文相关描述APUSIC_MQ_HOMEApusicMQ安装根目录1.
3参考文档金蝶Apusic消息中间件v9.
0用户手册第二章安装指南2.
1系统要求安装AMQ的最低系统要求见以下内容:环境要求操作系统Windows/Unix/Linux等主流操作系统Java环境Jdk1.
8内存1G以上硬盘200M以上推荐配置:环境要求内存2G以上硬盘10G以上2.
2安装AMQ产品采用java语言开发,产品完全支持跨平台安装.
产品包以zip压缩包的形式提供,只需配置java环境变量,安装方便简单.
2.
2.
1获取安装包Apusic-MQ-V9.
0-OS-Independent.
zip(Linux平台,Unix平台和Windows平台通用)可以从AMQ产品光盘中获得安装包文件.
2.
2.
2安装从光盘中获得安装包解压即安装.
2.
2.
3配置环境变量默认读取系统的jdk环境变量,也可在启动脚本(apusicmq.
bat或apusicmq)金蝶Apusic消息中间件v9.
0用户手册里指定,如下图:2.
3目录结构在执行完成AMQ的安装工作之后,在APUSIC_MQ_HOME目录下生成的目录及子目录如下图:具体描述如下表:子目录目录描述binAMQ运行脚本clientdemoAMQ客户端开发源码示例confAMQ相关配置文件dataAMQ持久化日志文件docAMQ开发与使用文档金蝶Apusic消息中间件v9.
0用户手册libAMQ核心jar包logAMQ运行日志文件spAMQ补丁包目录webappsAMQweb管理控制台应用目录license.
xmlAMQ授权文件2.
4卸载首先要停止AMQ服务;然后删除APUSIC_MQ_HOME目录即可.
金蝶Apusic消息中间件v9.
0用户手册第三章快速开始3.
1启动/停止服务运行脚本,在命令行提示符下,进入APUSIC_MQ_HOME/bin目录,3.
1.
1启动服务Windows:输入apusicmq.
batstart命令并执行.
Linux:输入apusicmqstart命令并执行(后台启动).
输入apusicmqconsole命令并执行(前台启动).
.
3.
1.
2停止服务Windows:输入apusicmq.
batstop命令并执行.
Linux:输入apusicmqstop命令并执行.
3.
2版本信息Windows:输入apusicmq.
bat--version命令并执行.
Linux:输入apusicmq--version命令并执行.
3.
3登陆管理控制台启动AMQ后,打开浏览器输入web管理控制台网址:默认是http://ip:8161/console,ip是安装AMQ服务器的ip地址.
完成后进入管理控制台登录页面,如下图金蝶Apusic消息中间件v9.
0用户手册在弹出的对话框中输入用户名和密码,默认是admin/admin,即可登录成功,进入首页,如下图:3.
4创建队列在管理控制台,点击队列菜单,输入新增的队列名,即可添加队列,如下图:金蝶Apusic消息中间件v9.
0用户手册3.
5发送消息在管理控制台,点击发送消息菜单,输入发送的内容即可发送消息,如下图:金蝶Apusic消息中间件v9.
0用户手册第四章ApusicMQ介绍4.
1ApusicMQ的原理ApusicMQ是基于JMS的消息中间件.
下面先介绍一下JMS的基本构件.
4.
1.
1JMS的基本构件4.
1.
1.
1连接工厂连接工厂是客户用来创建连接的对象,ApusicMQ提供的连接工厂类为:ApusicMQConnectionFactory.
4.
1.
1.
2会话JMSSession是生产和消费消息的一个单线程上下文.
会话用于创建消息生产者(producer)、消息消费者(consumer)和消息(message)等.
会话提供了一个事务性的上下文,在这个上下文中,一组发送和接收被组合到了一个原子操作中.
客户端用Session创建MessageProducer和MessageConsumer对象.
如果在Session关闭时,有一些消息已经被收到,但还没有被签收(acknowledged),那么,当消费者下次连接到相同的队列时,这些消息还会被再次接收.
4.
1.
1.
3目的地目的地是客户用来指定它生产的消息的目标和它消费的消息的来源的对象.
JMS1.
0.
2规范中定义了两种消息传递域:点对点(PTP)消息传递域和发布/订阅消息传递域.
在点对点消息传递域中,目的地被成为队列(queue);在发布/订阅消息传递域中,目的地被成为主题(topic).
点对点消息传递域的特点如下:每个消息只能有一个消费者.
消息的生产者和消费者之间没有时间上的相关性.
无论消费者在生产者发送消息的时候是否处于运行状态,它都可以提取消息.
发布/订阅消息传递域的特点如下:金蝶Apusic消息中间件v9.
0用户手册每个消息可以有多个消费者.
生产者和消费者之间有时间上的相关性.
订阅一个主题的消费者只能消费自它订阅之后发布的消息.
JMS规范允许客户创建持久订阅,这在一定程度上放松了时间上的相关性要求.
持久订阅允许消费者消费它在未处于激活状态时发送的消息.
4.
1.
1.
4生产者消息生产者是由会话创建的一个对象,用于把消息发送到一个目的地.
4.
1.
1.
5消息消费者消息消费者是由会话创建的一个对象,它用于接收发送到目的地的消息.
消息的消费可以采用以下两种方法之一:同步消费.
通过调用消费者的receive方法从目的地中显式提取消息.
receive方法可以一直阻塞到消息到达.
异步消费.
客户可以为消费者注册一个消息监听器,以定义在消息到达时所采取的动作.
4.
1.
1.
6消息JMS消息由以下三部分组成:消息头.
每个消息头字段都有相应的getter和setter方法.
消息属性.
如果需要除消息头字段以外的值,那么可以使用消息属性.
除了消息头中定义好的标准属性外,JMS提供一种机制增加新属性到消息头中,这种新属性包含以下几种:1.
应用需要用到的属性;2.
消息头中原有的一些可选属性;3.
JMSProvider需要用到的属性.
标准的JMS消息头包含以下属性:消息头描述JMSDestination消息发送的目的地JMSDeliveryMode传送模式,有两种模式:PERSISTENT和NON_PERSISTENT.
PERSISTENT表示该消息会进行持久化,一定要被送到目的地,保证可靠传输;NON_PERSISTENT表示该消息非持久化,但当AMQ服务停止后,传输中的消息会丢失.
这两种模式使开发者可以在消息传送的可靠性和吞吐量之间找到平衡点.
金蝶Apusic消息中间件v9.
0用户手册JMSExpiration消息过期时间,等于Destination的send方法中的timeToLive值加上发送时刻的GMT时间值.
如果timeToLive值等于零,则JMSExpiration被设为零,表示该消息永不过期.
如果发送后,在消息过期时间之后消息还没有被发送到目的地,则该消息被清除.
JMSPriority消息优先级,从0-9十个级别,0-4是普通消息,5-9是加急消息.
JMS不要求JMSProvider严格按照这十个优先级发送消息,但必须保证加急消息要先于普通消息到达.
JMSMessageID唯一识别每个消息的标识,由JMSProvider产生.
JMSTimestamp一个消息被提交给JMSProvider到消息被发出的时间.
JMSCorrelationID用来连接到另外一个消息,典型的应用是在回复消息中连接到原消息.
JMSReplyTo提供本消息回复消息的目的地址JMSType消息类型的识别符.
JMSRedelivered如果一个客户端收到一个设置了JMSRedelivered属性的消息,则表示可能客户端曾经在早些时候收到过该消息,但并没有签收(acknowledged).
消息体.
JMS定义的消息类型有五种:TextMessage、MapMessage、BytesMessage、StreamMessage、ObjectMessage.
4.
1.
2JMS的可靠性机制当所有的消息必须被接收,则采用持久化模式,但传输效率会打折扣.
当丢失消息能够被容忍,则用非持久模式,优点是传送速度快.
4.
1.
2.
1消息确认JMS消息只有在被确认之后,才认为已经被成功地消费了.
消息的成功消费通常包含三个阶段:客户接收消息、客户处理消息和消息被确认.
在事务性会话中,当一个事务被提交的时候,确认自动发生.
在非事务性会话中,消息何时被确认取决于创建会话时的应答模式(acknowledgementmode).
该参数有以下三个可选值:Session.
AUTO_ACKNOWLEDGE.
当客户成功的从receive方法返回的时候,或者从MessageListener.
onMessage方法成功返回的时候,会话自动确认客户收到的消息.
金蝶Apusic消息中间件v9.
0用户手册Session.
CLIENT_ACKNOWLEDGE.
客户通过消息的acknowledge方法确认消息.
需要注意的是,在这种模式中,确认是在会话层上进行:确认一个被消费的消息将自动确认所有已被会话消费的消息.
例如,如果一个消息消费者消费了10个消息,然后确认第5个消息,那么所有10个消息都被确认.
Session.
DUPS_ACKNOWLEDGE.
该选择只是会话迟钝第确认消息的提交.
如果JMSprovider失败,那么可能会导致一些重复的消息.
如果是重复的消息,那么JMSprovider必须把消息头的JMSRedelivered字段设置为true.
4.
1.
2.
2持久性JMS支持以下两种消息提交模式:PERSISTENT.
指示JMSprovider持久保存消息,以保证消息不会因为JMSprovider的失败而丢失.
NON_PERSISTENT.
不要求JMSprovider持久保存消息.
4.
1.
2.
3优先级可以使用消息优先级来指示JMSprovider首先提交紧急的消息.
优先级分10个级别,从0(最低)到9(最高).
如果不指定优先级,默认级别是4.
需要注意的是,JMSprovider并不一定保证按照优先级的顺序提交消息.
4.
1.
2.
4消息过期可以设置消息在一定时间后过期,默认是永不过期.
4.
1.
2.
5临时目的地可以通过会话上的createTemporaryQueue方法和createTemporaryTopic方法来创建临时目的地.
它们的存在时间只限于创建它们的连接所保持的时间.
只有创建该临时目的地的连接上的消息消费者才能够从临时目的地中提取消息.
4.
1.
2.
6持久订阅首先消息生产者必须使用PERSISTENT提交消息.
客户可以通过会话上的createDurableSubscriber方法来创建一个持久订阅,该方法的第一个金蝶Apusic消息中间件v9.
0用户手册参数必须是一个topic.
第二个参数是订阅的名称.
JMSprovider会存储发布到持久订阅对应的topic上的消息.
如果最初创建持久订阅的客户或者任何其它客户使用相同的连接工厂和连接的客户ID、相同的主题和相同的订阅名再次调用会话上的createDurableSubscriber方法,那么该持久订阅就会被激活.
JMSprovider会向客户发送客户处于非激活状态时所发布的消息.
持久订阅在某个时刻只能有一个激活的订阅者.
持久订阅在创建之后会一直保留,直到应用程序调用会话上的unsubscribe方法.
4.
1.
2.
7本地事务在一个JMS客户端,可以使用本地事务来组合消息的发送和接收.
JMSSession接口提供了commit和rollback方法.
事务提交意味着生产的所有消息被发送,消费的所有消息被确认;事务回滚意味着生产的所有消息被销毁,消费的所有消息被恢复并重新提交,除非它们已经过期.
事务性的会话总是牵涉到事务处理中,commit或rollback方法一旦被调用,一个事务就结束了,而另一个事务被开始.
关闭事务性会话将回滚其中的事务.
需要注意的是,如果使用请求/回复机制,即发送一个消息,同时希望在同一个事务中等待接收该消息的回复,那么程序将被挂起,因为知道事务提交,发送操作才会真正执行.
需要注意的还有一个,消息的生产和消费不能包含在同一个事务中.
4.
2ApusicMQ基本配置ApusicMQ配置文件:APUSIC_MQ_HOME/conf/apusicmq.
xml4.
2.
1配置ApusicMQ端口在transportConnectors标识中配置ApusicMQ服务端口,其中name属性指定协议的名称,uri属性指定协议所对应的协议名和端口号.
Java客户端默认使用openwire协议,所以ApusicMQ服务地址为tcp://ip:61616金蝶Apusic消息中间件v9.
0用户手册4.
2.
2监控ApusicMQ本节将使用JMX监控ApusicMQ.
……配置JMX步骤如下:1.
设置broker标识的useJmx属性为true;2.
取消对managementContext标识的注释(系统默认注释managementContext标识),监控的默认端口为1099.
在Windows平台监控进入%JAVA_HOME%/bin,双击jconsole.
exe,在对话框中输入ApusicMQ服务主机的地址,JMX的端口和主机登录帐号.
4.
3ApusicMQ的BrokerApusicMQ中,Broker是消息的代理者,消费者和生产者实际是通过Broker来进行消息的传递,生产者和消费者之间是无法直接通信的.
Broker的实现可以分为如下几种方式:金蝶Apusic消息中间件v9.
0用户手册4.
3.
1RunningBroker:ApusicMQ产品包中bin目录中包含一个名为apusicmq的脚本,直接运行这个脚本就可以启动一个broker.
此外也可以通过BrokerConfigurationURI或BrokerXBeanURI对broker进行配置,以下是一些命令行参数的例子:示例描述apusicmqstart运行一个broker使用默认的apusicmq.
xml配置文件apusicmqstartxbean:myconfig.
xml运行一个broker使用当前路径下的myconfig.
xml配置文件apusicmqstartxbean:file:.
/conf/broker1.
xml运行一个broker使用broker1.
xml作为配置文件使用相对路径.
/conf/broker1.
xmlapusicmqstartxbean:file:C:/ApusicMQ/conf/broker2.
xml运行一个broker使用broker2.
xml作为配置文件使用绝对路径C:/ApusicMQ/conf/broker2.
xml4.
3.
2EmbeddedBroker:可以通过在应用程序中以编码的方式启动broker,例如:Java代码BrokerServicebroker=newBrokerService();broker.
addConnector("tcp://localhost:61616");broker.
start();如果需要启动多个broker,那么需要为broker设置一个名字.
例如:Java代码BrokerServicebroker=newBrokerService();broker.
setName("fred");broker.
addConnector("tcp://localhost:61616");broker.
start();如果希望在同一个JVM内访问这个broker,那么可以使用VMTransport,URI是:vm://brokerName.
此外,也可以通过BrokerFactory来创建broker,例如:Java代码金蝶Apusic消息中间件v9.
0用户手册BrokerServicebroker=BrokerFactory.
createBroker(newURI(someURI));someURI的可选值如下:参数示例描述xbean:xbean:ApusicMQ.
xml查找运行环境(classpath)相对路径下的配置文件(如果在根目录下配置为---xbean:ApusicMQ.
xml)file:file:conf/ApusicMQ.
xml查找实例目录的broker配置文件(conf/ApusicMQ.
xml)broker:broker:tcp://localhost:61616通过URL配置当使用XBean的配置方式的时候,需要指定一个xml配置文件,例如:Java代码BrokerServicebroker=BrokerFactory.
createBroker(newURI("xbean:com/test/ApusicMQ.
xml"));使用Spring的配置方式如下:Xml代码4.
4ApusicMQBroker的TransportTransport是Broker数据传输使用的协议,ApusicMQ目前支持的transport有:TCPTransport、SSLTransport、FailoverTransport、DiscoveryTransport、等.
下面介绍几种常用的Transport金蝶Apusic消息中间件v9.
0用户手册4.
4.
1TCPTransportTCPtransport允许客户端通过TCPsocket连接到远程的broker.
以下是配置语法:tcp://hostname:porttransportOptionsTransportOptions的可选值如下:参数默认值描述minmumWireFormatVersion0最低版本wireformat是允许的.
tracefalse用于打开日志,记录每次的通过该链接通道的命令,true为记录.
useLocalHosttrue当设置为true时,本机名字"localhost".
socketBufferSize64*1024设置socket缓存大小,单位byte.
soTimeout0设置socket超时时间,单位毫秒.
connectionTimeout30000非零值指定连接超时为毫秒.
零值意味着永远等待连接建立.
负值被忽略.
wireFormatdefault使用的wireformat名称.
wireFormat.
*所有与这个前缀的属性用于配置wireformat.
有关更多信息,请参阅配置有线格式4.
4.
2FailoverTransportFailoverTransport是一种重新连接的机制,它工作于其它transport的上层,用于建立可靠的传输.
它的配置语法允许制定任意多个复合的URI.
Failovertransport会自动选择其中的一个URI来尝试建立连接.
如果没有成功,那么会选择一个其它的URI来建立一个新的连接.
以下是配置语法:failover:(uri1,.
.
.
,uriN)transportOptionsTransportOptions的可选值如下:参数默认值描述initialReconnectDelay10在第一次重新连接尝试之前要等待多长时间(毫秒).
maxReconnectDelay30000重新连接尝试之间等待的最大时间(毫秒).
useExponentialBackOfftrue启用指数倍数递增的方式增加延迟时间.
金蝶Apusic消息中间件v9.
0用户手册该指数退避用于连接尝试之间reconnectDelayExponent2.
0TheexponentusedintheexponentialbackoffattemptsmaxReconnectAttempts-1|0从版本5.
6开始:-1是默认的,永远意味着重试,0意味着不要重试(只尝试连接一次,但不重试).
在版本5.
6之前:0是默认的,永远意味着重试.
所有版本:如果设置为>0,那么这是在将错误发送回客户端之前重新连接尝试的最大次数.
startupMaxReconnectAttempts0如果不是0,那么这是一个错误是在发送回客户端的客户端第一次尝试开始重新尝试连接的最大数目,一旦连接maxreconnectattempts选项优先.
randomizetrue使用随机算法选择用于从所提供的列表中重新连接的URI.
backupfalse初始化并保存第二个传输连接-启用快速故障转移.
timeout-1使超时发送操作(以毫秒)不重过程中断.
trackMessagesfalse保存一个缓存的消息,这些消息将在重新连接时刷新到代理.
maxCacheSize131072用于缓存的大小,如果trackmessages启用.
updateURIsSupportedtrue确认客户是否应该接受更新其已知的URI列表从连接的经纪人.
在apusicmq5.
4加.
updateURIsURLnull一个URL(或路径到本地文件)到一个包含一个以逗号分隔的列表URI用于在失败的情况下重新连接的文本文件.
在apusicmq5.
4加.
nested.
*null添加到嵌套URL的额外选项.
在apusicmq5.
9加.
warnAfterReconnectAttempts.
*10每次重新连接后,尝试记录一个警告,指示没有连接,但仍在尝试,设置为金蝶Apusic消息中间件v9.
0用户手册.
.
.
在使用FailoverTransport或Discoverytransport等能够自动重连的transport的时候,需要注意的是:设想有两个broker,它们都启用AMQMessageStore作为持久化存储,有一个producer和一个consumer连接到某个queue.
当因其中一个broker失效时而切换到另一个broker的时候,如果失效的broker的queue中还有未被consumer消费的消息,那么这个queue里的消息仍然滞留在失效broker的中,直到失效的broker被修复并重新切换回这个被修复的broker后,之前被保留的消息才会被consumer消费掉.
如果被处理的消息有时序限制,那么应用程序就需要处理这个问题.
另外也可以通过ApusicMQ集群来解决这个问题.
在transport重连的时候,可以在connection上注册TransportListener来获得回调,例如:Java代码(ApusicMQConnection)connection).
addTransportListener(newTransportListener(){publicvoidonCommand(Objectcmd){}publicvoidonException(IOExceptionexp){}publicvoidtransportInterupted(){//Thetransporthassufferedaninterruptionfromwhichithopestorecover.
}publicvoidtransportResumed(){//Thetransporthasresumedafteraninterruption.
}});4.
5ApusicMQBroker的持久方式当ApusicMQ的队列比较多时,可能内存将会不够用,或突然关机,可能导致内存数据丢失,所以我们需要通过其他方式来保证数据的完整性.
目前Broker常用的持久方式有:AMQMessageStore,KahaPersistence,JDBCPersistence金蝶Apusic消息中间件v9.
0用户手册4.
5.
1KahaMessageStoreKahaPersistence是一个专门针对消息持久化的解决方案.
它对典型的消息使用模式进行了优化.
在Kaha中,数据被追加到datalogs中.
当不再需要log文件中的数据的时候,log文件会被丢弃.
以下是其配置的一个例子:Xml代码4.
5.
2JDBCMessageStore目前支持的数据库有DB2,HSQL,Informix,MySQL,Oracle,Postgresql,SQLServer,Sybase等.
如果你使用的数据库不被支持,那么可以调整StatementProvider来保证使用正确的SQL方言.
通常绝大多数数据库支持以下adaptor:com.
apusic.
amq.
store.
jdbc.
adapter.
BlobJDBCAdaptercom.
apusic.
amq.
store.
jdbc.
adapter.
BytesJDBCAdaptercom.
apusic.
amq.
store.
jdbc.
adapter.
DefaultJDBCAdaptercom.
apusic.
amq.
store.
jdbc.
adapter.
ImageJDBCAdapter也可以在配置文件中直接指定JDBCadaptor,例如:Xml代码1.
以下是其配置的一个例子:金蝶Apusic消息中间件v9.
0用户手册Xml代码4.
6ActimveMQ的其他特性4.
6.
1异步发送消息ApusicMQ支持生产者以同步或异步模式发送消息.
使用不同的模式对send方法的反应时间有巨大的影响,反映时间是衡量ApusicMQ吞吐量的重要因素,使用异步发送可以提高系统的性能.
在默认大多数情况下,ApusicMQ是以异步模式发送消息.
例外的情况:在没有使用事务的情况下,生产者以PERSISTENT传送模式发送消息.
在这种情况下,send方法都是同步的,并且一直阻塞直到ApusicMQ发回确认消息:消息已经存储在持久性数据存储中.
这种确认机制保证消息不会丢失,但会造成生产者阻塞从而影响反应时间.
高性能的程序一般都能容忍在故障情况下丢失少量数据.
如果编写这样的程序,可以通过使用异步发送来提高吞吐量(甚至在使用PERSISTENT传送模式的情况下).
金蝶Apusic消息中间件v9.
0用户手册使用ConnectionURI配置异步发送:cf=newApusicMQConnectionFactory("tcp://locahost:61616jms.
useAsyncSend=true");在ConnectionFactory层面配置异步发送:((ApusicMQConnectionFactory)connectionFactory).
setUseAsyncSend(true);在Connection层面配置异步发送,此层面的设置将覆盖ConnectionFactory层面的设置:((ApusicMQConnection)connection).
setUseAsyncSend(true);4.
6.
2消费者异步分派在ApusicMQ中,支持ApusicMQ以同步或异步模式向消费者分派消息.
这样的意义:可以以异步模式向处理消息慢的消费者分配消息;以同步模式向处理消息快的消费者分配消息.
ApusicMQ默认以同步模式分派消息,这样的设置可以提高性能.
但是对于处理消息慢的消费者,需要以异步模式分派.
在ConnectionFactory层面配置同步分派:((ApusicMQConnectionFactory)connectionFactory).
setDispatchAsync(false);在Connection层面配置同步分派,此层面的设置将覆盖ConnectionFactory层面的设置:((ApusicMQConnection)connection).
setDispatchAsync(false);在消费者层面以DestinationURI配置同步分派,此层面的设置将覆盖ConnectionFactory和Connection层面的设置:queue=newApusicMQQueue("TEST.
QUEUEconsumer.
dispatchAsync=false");consumer=session.
createConsumer(queue);4.
6.
3消费者优先级在ApusicMQ分布式环境中,在有消费者存在的情况下,如果更希望ApusicMQ发送消息给消费者而不是其他的ApusicMQ到ApusicMQ的传送,可以如下设置:Java客户端:queue=newApusicMQQueue("TEST.
QUEUEconsumer.
prority=10");consumer=session.
createConsumer(queue);金蝶Apusic消息中间件v9.
0用户手册4.
6.
4独占消费者ApusicMQ维护队列消息的顺序并顺序把消息分派给消费者.
但是如果建立了多个Session和MessageConsumer,那么同一时刻多个线程同时从一个队列中接收消息时就并不能保证处理时有序.
有时候有序处理消息是非常重要的.
ApusicMQ支持独占的消费.
ApusicMQ挑选一个MessageConsumer,并把一个队列中所有消息按顺序分派给它.
如果消费者发生故障,那么ApusicMQ将自动故障转移并选择另一个消费者.
可以如下设置:Java客户端:queue=newApusicMQQueue("TEST.
QUEUEconsumer.
exclusive=true");consumer=session.
createConsumer(queue);4.
6.
5消息重发策略在事务控制里抛出异常,txManager会进行rollback处理(在ApusicMQ里,消息默认会redelivery到客户端6次,如果继续异常,消息会放到deadletterqueue里(ApusicMQ.
DLQ)).
在以下三种情况中,消息会被再次传送给消费者:1.
在使用事务的Session中,调用rollback()方法;2.
在使用事务的Session中,调用commit()方法之前就关闭了Session;3.
在Session中使用CLIENT_ACKNOWLEDGE签收模式,并且调用了recover()方法.
可以通过设置ApusicMQConnectionFactory和ApusicMQConnection来定制想要的再次传送策略.
你可以在ApusicMQConnectionFactory或ApusicMQConnection类中配置RedeliveryPolicy属性,用来定义重传策略的具体细节.
可以使用Java代码,Spring配置或配置URI字符串来定义重传策略.
可用的属性:属性默认值说明collisionAvoidanceFactor0.
15设置防止冲突范围的正负百分比,只有启用useCollisionAvoidance参数时才生效.
maximumRedeliveries6最大重传次数,达到最大重连次数后抛出异常.
为-1时不限制次数,为0时表示不进行重传.
金蝶Apusic消息中间件v9.
0用户手册maximumRedeliveryDelay-1最大传送延迟,只在useExponentialBackOff为true时有效(V5.
5),假设首次重连间隔为10ms,倍数为2,那么第二次重连时间间隔为20ms,第三次重连时间间隔为40ms,当重连时间间隔大的最大重连时间间隔时,以后每次重连时间间隔都为最大重连时间间隔.
initialRedeliveryDelay1000L初始重发延迟时间redeliveryDelay1000L重发延迟时间,当initialRedeliveryDelay=0时生效(v5.
4)useCollisionAvoidancefalse启用防止冲突功能,因为消息接收时是可以使用多线程并发处理的,应该是为了重发的安全性,避开所有并发线程都在同一个时间点进行消息接收处理.
所有线程在同一个时间点处理时会发生什么问题呢应该没有问题,只是为了平衡broker处理性能,不会有时很忙,有时很空闲.
useExponentialBackOfffalse启用指数倍数递增的方式增加延迟时间.
backOffMultiplier5重连时间间隔递增倍数,只有值大于1和启用useExponentialBackOff参数时才生效4.
6.
6目标相关的属性ApusicMQ支持混合目标技术.
它允许在一个JMS目标中使用一组JMS目标.
例如可以利用混合目标在同一操作中用向12个队列发送同一条消息或者在同一操作中向一个主题和一个队列发送同一条消息.
在混合目标中,通过","来分隔不同的目标.
Java客户端:例如://sendto3queuesasonelogicaloperationQueuequeue=newApusicMQQueue("FOO.
A,FOO.
B,FOO.
C");producer.
send(queue,someMessage);如果在一个目标中混合不同类别的目标,可以通过使用"queue://"和"topic://"前缀来识别不同的目标.
金蝶Apusic消息中间件v9.
0用户手册例如://sendtoqueuesandtopiconelogicaloperationQueuequeue=newApusicMQQueue("FOO.
A,topic://NOTIFY.
FOO.
A");producer.
send(queue,someMessage);目标选项有:属性默认值描述consumer.
prefetchSizevariable消费者将预取的消息数consumer.
maximumPendingMessageLimit0用于控制如果存在缓慢的消费者情况下是否删除消息x.
consumer.
noLocalFALSE相同主题的消费者那样的旗帜.
在这里公开,以便它可以与队列一起使用.
consumer.
dispatchAsyncFALSE代理是否向用户异步发送消息consumer.
retroactiveFALSE是否追溯消费者consumer.
selectornull与消费者一起使用的JMS选择器consumer.
exclusiveFALSE是否是独家消费者consumer.
priority0允许您配置消费者优先级Java客户端的使用:例如:queue=newApusicMQQueue("TEST.
QUEUEconsumer.
dispatchAsync=false&consumer.
prefetchSize=10");consumer=session.
createConsumer(queue);4.
6.
7消息预取处理ApusicMQ的目标之一就是高性能的数据传送,所以ApusicMQ使用"预取限制"来控制有多少消息能及时的传送给任何地方的消费者.
一旦预取数量达到限制,那么就不会有消息被分派给这个消费者直到它发回签收消息(用来标识所有的消息已经被处理).
金蝶Apusic消息中间件v9.
0用户手册可以为每个消费者指定消息预取.
如果有大量的消息并且希望更高的性能,那么可以为这个消费者增大预取值.
如果有少量的消息并且每条消息的处理都要花费很长的时间,那么可以设置预取值为1,这样同一时间,ApusicMQ只会为这个消费者分派一条消息.
Java客户端:在ConnectionFactory层面为所有消费者配置预取值:tcp://localhost:61616jms.
prefetchPolicy.
all=50在ConnectionFactory层面为队列消费者配置预取值:tcp://localhost:61616jms.
prefetchPolicy.
queuePrefetch=1使用"目标选项"为一个消费者配置预取值:queue=newApusicMQQueue("TEST.
QUEUEconsumer.
prefetchSize=10");consumer=session.
createConsumer(queue);4.
6.
8配置连接URLApusicMQ支持通过ConfigurationURI明确的配置连接属性.
例如:当要设置异步发送时,可以通过在ConfigurationURI中使用jms.
$PROPERTY来设置.
tcp://localhost:61616jms.
useAsyncSend=true以下的选项在URI必须以"jms.
"为前缀.
属性默认值描述alwaysSessionAsyncTRUE如果设置了此标志,则不使用单独的线程来为连接中的每个会话调度消息.
然而,一个单独的线程总是如果有一个以上的会议,或会议不在自动应答或DUPSOK模式clientIDnull设置JMSclientId用于连接closeTimeout15000(milliseconds)设置关闭之前被认为是完整的结束.
通常,一个连接等待broker确认;这允许操作超时保存客户端挂如果没有broker.
copyMessageOnSendTRUE是否复制JMS消息一个新的JMS消息对象在JMS的send方法部分.
默认情况下,这是符合JMS规范的.
如果在发送性能提升后不修改JMS消息,可以禁用它.
disableTimeStampsByDefaultFALSE组上的时间戳信息是否禁用或不.
如果禁用它们,它会增加一个小的性能提升.
金蝶Apusic消息中间件v9.
0用户手册dispatchAsyncFALSE代理是否向用户异步发送消息nestedMapAndListEnabledTRUE启用/禁用结构化信息的性质和mapmessages支持使消息属性和mapmessage条目可以包含嵌套的地图和列表对象.
从4.
1版起可用objectMessageSerializationDeferedFALSE当一个对象放在一个objectmessage,JMS规范要求的对象进行序列化的设置方法.
启用此标志将使对象无法序列化.
如果消息需要通过套接字发送或存储到磁盘,则对象可能随后序列化.
optimizeAcknowledgeFALSE启用一种优化的确认模式,在该模式中,消息是分批确认的,而不是单独的.
或者,你可以使用session.
dups_ok_acknowledge确认模式,往往可以更快的消费者.
启用此问题的警告可能会导致重新连接时自动确认的一些问题.
optimizedMessageDispatchTRUE如果设置了此标志,则使用较大的预取限制—仅适用于持久主题订阅者.
useAsyncSendFALSE用异步发送增加了巨大的性能提升;但意味着send方法会立即返回消息是否已经发送或不可能导致信息丢失.
useCompressionFALSE允许使用消息体压缩useRetroactiveConsumerFALSE设置是否启用追溯消费者.
追溯用户允许非持久性主题订阅者接收在非持久订户启动之前发布的旧消息.
4.
6.
9消息重发与死信管理DLQ-死信队列(DeadLetterQueue)用来保存处理失败或者过期的消息.
出现以下情况时,消息会被redeliveredAtransactedsessionisusedandrollback()iscalled.
Atransactedsessionisclosedbeforecommitiscalled.
AsessionisusingCLIENT_ACKNOWLEDGEandSession.
recover()iscalled.
当一个消息被redelivered超过maximumRedeliveries(缺省为6次,具体设置请参考后面的链接)次数时,会给broker发送一个"Poisonack",这个消息被认为是apoisonpill,这时broker会将这个消息发送到DLQ,以便后续处理.
缺省的死信队列是ApusicMQ.
DLQ,如果没有特别指定,死信都会被发送到这个队列.
缺省持久消息过期,会被送到DLQ,非持久消息不会送到DLQ可以通过配置文件(apusicmq.
xml)来调整死信发送策略.
1.
不使用缺省的死信队列金蝶Apusic消息中间件v9.
0用户手册缺省所有队列的死信消息都被发送到同一个缺省死信队列,不便于管理.
可以通过individualDeadLetterStrategy或sharedDeadLetterStrategy策略来进行修改.
如下:',否则用队列名称-->">.
.
.
2.
非持久消息保存到死信队列">3.
过期消息不保存到死信队列">金蝶Apusic消息中间件v9.
0用户手册4.
6.
10生产者和消费者的密码使用打开conf/apusicmq.
xml文件,在标签里的标签前加入:注意必须在标签前,否则启动ApusicMQ会报错.
金蝶Apusic消息中间件v9.
0用户手册第五章ApusicMQ集群方式5.
1AMQ集群特点异构系统通讯、异步调用、应用解耦.
高可用性.
每组Broker采用主从模式部署,支持Failover.
客户端支持多组Broker服务器负载均衡和Failover.
统一的异常消息重试服务,便于容错.
同步发送、消息持久化,防止数据丢失.
消息异步归档到云存储,便于问题跟踪.
高扩展性.
动态扩容,支持水平扩展,按消息垂直划分.
易维护.
统一集群管理控制平台,完善的监控报警,消息运营状况统计.
原理图:5.
2AMQ集群方式--QueueconsumerclustersApusicMQ支持订阅同一个queue的consumers上的集群.
如果一个consumer失效,那么所有未被确认(unacknowledged)的消息都会被发送到这个queue上其它的consumers.
如果某个consumer的处理速度比其它consumers更快,那么这个consumer就会消费更多的消息.
金蝶Apusic消息中间件v9.
0用户手册5.
3AMQ集群方式--Brokerclusters一个常见的场景是有多个JMSbroker,有一个客户连接到其中一个broker.
如果这个broker失效,那么客户会自动重新连接到其它的broker.
在ApusicMQ中使用failover://协议来实现这个功能.
ApusicMQ3.
x版本的reliable://协议已经变更为failover://.
如果某个网络上有多个brokers而且客户使用静态发现(使用StaticTransport或FailoverTransport)或动态发现(使用DiscoveryTransport),那么客户可以容易地在某个broker失效的情况下切换到其它的brokers.
然而,standalonebrokers并不了解其它brokers上的consumers,也就是说如果某个broker上没有consumers,那么这个broker上的消息可能会因得不到处理而积压起来.
目前的解决方案是使用Networkofbrokers,以便在broker之间存储转发消息.
ApusicMQ支持networksofbrokers.
它支持分布式的queues和topics.
一个broker会相同对待所有的订阅(subscription):不管他们是来自本地的客户连接,还是来自远程broker,它都会递送有关的消息拷贝到每个订阅.
远程broker得到这个消息拷贝后,会依次把它递送到其内部的本地连接上.
有两种方式配置Networkofbrokers,一种是使用statictransport,如下:金蝶Apusic消息中间件v9.
0用户手册…另外一种是使用multicastdiscovery,如下:.
.
.
NetworkConnector有以下属性:属性默认值描述namebridge名称dynamicOnlyfalse如果为true,持久订阅被激活时才创建对应的网路持久订阅.
默认是启动时激活.
decreaseNetworkConsumerPriorityfalse如果为true,网络的消费者优先级降低为-5.
如果为false,则默认跟本地消费者一样为0.
networkTTL1消息和订阅在网络上通过的broker数量conduitSubscriptionstrue多个网络消费者是否被当做一个金蝶Apusic消息中间件v9.
0用户手册消费者来对待.
excludedDestinationsempty不通过网络转发的destinationdynamicallyIncludedDestinationsempty通过网络转发的destinations,注意空列表代表所有的都转发.
staticallyIncludedDestinationsempty匹配的都将通过网络转发-即使没有对应的消费者duplexfalse如果为true,则既可消费又可生产消息到网络brokerprefetchSize1000设置网络消费者的prefetchsize参数.
必须大于0,因为网络消费者不能自己轮询消息.
suppressDuplicateQueueSubscriptionsfalse(从5.
3版本开始)如果为true,重复的订阅关系一产生即被阻止.
bridgeTempDestinationstrue是否广播advisorymessages来创建临时destination.
alwaysSyncSendfalse(从5.
6版本开始)如果为true,非持久化消息也将使用request/reply方式代替oneway方式发送到远程broker.
staticBridgefalse(从5.
6版本开始)如果为true,只有staticallyIncludedDestinations中配置的destination可以被处理.
5.
3.
1管道订阅conduitSubscriptions对于conduitSubscriptions属性,用于负载均衡时使用.
设想有两个brokers,分别是brokerA和brokerB,它们之间用forwardingbridge连接.
有一个consumer连接到brokerA并订阅queue:Q.
TEST.
有两个consumers连接到brokerB,也是订阅queue:Q.
TEST.
这三个consumers有相同的优先级.
然后启动一个producer,它发送了30条消息到brokerA.
如果conduitSubscriptions=true,那么brokerA上的consumer会得到15条消息,另外15条消息会发送给brokerB.
此时负载并不均衡,因为此时brokerA将brokerB上的两个consumers视为一个;如果conduitSubscriptions=false,那么每个consumer上都会收到10条消息.
5.
3.
2双向网络连接(duplexnetworkConnector)默认NetworkConnector在需要转发消息时是单向连接的.
当duplex=true时,就变成了双向连接,这时配置在broker2端的指向broker1的duplexnetworkConnector,相当于即配置了金蝶Apusic消息中间件v9.
0用户手册broker2到broker1的网络连接,也配置了broker1到broker2的网络连接.
(就是说不管broker1同意与否,都被绑架了.
)当然,仅仅在broker1上配置也有同样的效果.
注意:可以在两个broker间建立两个以上的双向网络连接来增加吞吐量或对主题\队列分区,只需要指定他们使用不同的name即可.
5.
3.
3指定和限制Destination通过NetworkConnector共享的destination太多,传输的AdvisoryMessage就会变的非常多,系统的拓扑结构将变得非常复杂,所有才有多种方式来限制这些destination配置:dynamicallyIncludedDestinations这里匹配到的destination,在需要时将被转发.
excludedDestinations这里匹配到的destination,将不会被转发staticallyIncludedDestinations如果指定了staticBridge为true,则只有这里匹配的destination可以被转发.
此时本地broker完全被其他broker代理.
并且本broker不会订阅其他broker上的AdvisoryMessage,也不会获取任何远程consumer信息.
这几个配置可以使用通配符,比如">",详见wildcard.
示例代码:金蝶Apusic消息中间件v9.
0用户手册此外,从5.
6版本起,可以在networkConnector上设置destinationFilter来指定感兴趣的AdvisoryMessage将被传播.
5.
3.
4被卡住的消息一个很有意思的场景是,broker1和broker2通过networkConnector连接.
一些个consumers连接到broker1,消费broker2上的消息.
消息先被broker1从broker2上消费掉,然后转发给这些consumers.
不幸的是转发部分消息的时候broker1重启了,这些consumers发现broker1连接失败,通过failover连接到broker2上去了,但是有一部分他们还没有消费的消息被broker2已经分发到了broker1上去了.
这些消息,就好像是消失了,除非有消费者重新连接到broker1上来消费.
怎么办呢办法就是从5.
6版本destinationPolicy上新增的选项replayWhenNoConsumers.
这个选项使得broker1上有需要转发的消息但是没有消费者时,把消息回流到它原始的broker.
同时把enableAudit设置为false,为了防止消息回流后被当做重复消息而不被分发.
金蝶Apusic消息中间件v9.
0用户手册">5.
3.
5其他说明.
NetworkConnector基于AdvisoryMessage机制,如果broker的advisorySupport选型被禁用,则NetworkConnector将不起作用.
用作转发的broker中入列出列这些统计信息只记录其转发的数据.
用作转发的broker中无法看到远程broker的相同队列中的数据(browse消息列表为空,queuesize为0).
5.
4AMQ集群方式--MasterSlave在一个网络内运行多个brokers或者standalonebrokers时存在一个问题,这就是消息在物理上只被一个broker持有,因此当某个broker失效,那么你只能等待直到它重启后,这个broker上的消息才能够被继续发送(如果没有设置持久化,那么在这种情况下,消息将会丢失).
MasterSlave背后的想法是,消息被复制到slavebroker,因此即使masterbroker遇到了像硬件故障之类的错误,你也可以立即切换到slavebroker而不丢失任何消息.
MasterSlave是目前ApusicMQ推荐的高可靠性和容错的解决方案.
以下是几种不同的类型:MasterSlaveTypeRequirementsProsConsPureMasterSlaveNone无中心故障点需要手动重新启动才能返回失败的主机,并且只能支持1个从机MasterSlaveTypeRequirementsProsConsSharedFileSystemMasterSlaveASharedFilesystemsuchasaSAN之前主机的自动恢复需要共享文件系统金蝶Apusic消息中间件v9.
0用户手册JDBCMasterSlaveAShareddatabase按规定跑多少从机.
之前主机的自动恢复需要共享数据库.
也相对较慢,因为它不能使用高性能期刊.
5.
4.
1PureMasterSlavePureMasterSlave该方式已经逐渐被淘汰,这里不再描述.
5.
4.
2SharedFileSystemMasterSlave如果你使用SAN或者共享文件系统,那么你可以使用SharedFileSystemMasterSlave.
基本上,你可以运行多个broker,这些broker共享数据目录.
当第一个broker得到文件上的排他锁之后,其它的broker便会在循环中等待获得这把锁.
客户端使用failovertransport来连接到可用的broker.
当masterbroker失效的时候会释放这把锁,这时候其中一个slavebroker会得到这把锁从而成为masterbroker.
以下是ApusicMQ配置的一个例子:…5.
4.
3JDBCMasterSlaveJDBCMasterSlave的工作原理跟SharedFileSystemMasterSlave类似,只是采用了数据库作为持久化存储.
以下是ApusicMQ配置的一个例子:.
.
.
金蝶Apusic消息中间件v9.
0用户手册金蝶Apusic消息中间件v9.
0用户手册第六章ApusicMQ的JMX监控为了实时查看MQ当前的状况可以通过JMX来查看当前的情况,查找当前能够通过JMX获取的信息.
6.
1JMX的配置在启动脚本的APUSICMQ_SUNJMX_START参数中设置远程JMX连接的主机ip和端口,如:-Djava.
rmi.
server.
hostname=192.
168.
1.
12-Dcom.
sun.
management.
jmxremote.
port=10996.
2JMX的JAVA端获取信息//设置JMX服务地址JMXServiceURLurl=newJMXServiceURL(jmxUrl);JMXConnectorjmxc=JMXConnectorFactory.
connect(url,null);//获取JMX的连接MBeanServerConnectionserverConnection=jmxc.
getMBeanServerConnection();//设置相应的broker对象ObjectNamename=newObjectName("com.
apusic.
amq:type=Broker,brokerName="+this.
brokerName);//获取brokerVIEWBrokerViewMBeanmbean=(BrokerViewMBean)MBeanServerInvocationHandler.
newProxyInstance(serverConnection,name,BrokerViewMBean.
class,true);mo.
setBrokerName(mbean.
getBrokerName());//Topic信息ObjectName[]topics=mbean.
getTopics();ObjectName[]topicps=mbean.
getTopicProducers();ObjectName[]topicSubs=mbean.
getTopicSubscribers();//Queue信息ObjectName[]queues=mbean.
getQueues();ObjectName[]queuePs=mbean.
getQueueProducers();ObjectName[]queueSubs=mbean.
getQueueSubscribers();获取的这些ObjectName可以通过MBeanServerInvocationHandler.
newProxyInstance(serverConnection,objectName,class,true)来获取相应的详细信息,如:TopicViewMBean,QueueViewMBean,ProducerViewMBean,SubscriptionViewMBean金蝶Apusic消息中间件v9.
0用户手册