Jms之ActiveMQ学习笔记

ActiveMQ 是一个完全支持JMS1.1和J2EE 1.4规范的 JMS Provider实现,下载地址：http://activemq.apache.org/download.html

 

使用：

启动activeMQ，打开两个命令窗口，进入D:\activemq\example，一个运行：ant consumer,一个运行：ant producer，如果成功发送/接收到消息就可以了。

有的时候由于端口被占用,导致ActiveMQ错误,ActiveMQ可能需要以下端口1099(JMX),61616(默认的TransportConnector)

运行消息管理系统：http://localhost:8161/admin

 

连接方式：

ActiveMQConnectionFactory 提供了多种连接到Broker的方式，常见的有

vm://host:port

tcp://host:port

ssl://host:port

stomp://host:port  //stomp协议可以跨语言,目前有很多种stomp client 库(java,c#,c/c++,ruby,python...);

 

Broker:

1.Running Broker

直接运行bin目录下activemq.bat的脚本就可以启动一个broker

 

2.Embedded Broker

以编码的方式启动broker:

BrokerService broker = new BrokerService();

broker.setName("xxx");  //如果需要启动多个broker，那么需要为broker设置一个名字

broker.addConnector("tcp://localhost:61616");

broker.start();

如果希望在同一个JVM内访问这个broker，那么可以使用VM Transport，URI是：vm://brokerName

 

可通过BrokerService创建broker:

BrokerService broker = BrokerFactory.createBroker(new URI(uri));

uri可以使用以下形式：

xbean:activemq.xml

file:xx/activemq.xml

broker:tcp://localhost:61616

 

spring配置可以如下：

<bean id="broker" class="org.apache.activemq.xbean.BrokerFactoryBean">

<property name="config" value="classpath:xx/activemq.xml" />

<property name="start" value="true" />

</bean>

 

3.Monitoring Broker

 (1)jconsole:JMX(监控broker)

<broker xmlns="http://activemq.org/config/1.0" brokerName="localhost" useJmx="true">

<managementContext>

<!--止ActiveMQ创建自己的connector，可以为为JMX Connector配置密码保护-->

<managementContext createConnector="false"/>

</managementContext>

</broker>

启动jdk bin目录下jconsole.exe，进行相关的本地远程配置后就可以执行监控.

 (2)Web Console

直接访问http://localhost:8161/admin即可，可通过修改配置中的port属性修改端口

<!-- An embedded servlet engine for serving up the Admin console -->

<jetty xmlns="http://mortbay.com/schemas/jetty/1.0">

<connectors>

<nioConnector port="8161"/>

</connectors>

<handlers>

<webAppContext contextPath="/admin" resourceBase="${activemq.base}/webapps/admin" logUrlOnStart="true"/>

......

</handlers>

</jetty>

 (3)Advisory Message(通过JMS消息来监控系统)

通过它，你可以得到可以得到关于JMS provider、producers、consumers和destinations的一系列信息，

如生产者，消费者启停Queue或Topic消息的信息，Queue或Topic创建或销毁的信息，过期消息等。

Destination destination = AdvisorySupport.getProducerAdvisoryTopic(destination)

MessageConsumer consumer = session.createConsumer(destination);

consumer.setMessageListener(this);

Client advisories:

ActiveMQ.Advisory.Connection

ActiveMQ.Advisory.Producer.Queue(/Topic)...

ActiveMQ.Advisory.Consumer.Queue(/Topic)...

Destination advisories

ActiveMQ.Advisory.Queue(/Topic)...

ActiveMQ.Advisory.TempQueue(/TempTopic)...

ActiveMQ.Advisory.Expired.Queue(/Topic)...

ActiveMQ.Advisory.NoConsumer.Queue(/Topic)...

 (4)Command Agent

XMPP(Jabber)协议是一种基于XML的即时通信协议

<transportConnectors>

.....

<transportConnector name="xmpp" uri="xmpp://localhost:61222"/>

</transportConnectors>

 

启用Command Agent

<beans>

<broker useJmx="true" xmlns="http://activemq.org/config/1.0">

...

</broker>

<commandAgent xmlns="http://activemq.org/config/1.0"

　brokerUser="user" brokerPassword="passward"/>

</beans>

启用了Command Agent的broker上会有一个来自Command Agent的连接，它同时订阅topic：ActiveMQ.Agent;在你启动XMPP客户端，

加入到ActiveMQ.Agent聊天室后，就可以同broker进行交谈了。通过在XMPP 客户端中键入help，可以得到帮助信息

 (5)图表信息描述

ActiveMQ支持以broker插件的形式生成dot文件(可以用agrviewer来查看)，以图表的方式描述connections、sessions、producers、consumers、destinations等信息

<broker xmlns="http://activemq.org/config/1.0" brokerName="localhost" useJmx="true">

...

<plugins>

<connectionDotFilePlugin  file="connection.dot"/>

<destinationDotFilePlugin file="destination.dot"/>

</plugins>

</broker>

 

消息发送方式:

ActiveMQ支持以同步（sync）方式或者异步（async）方式向broker发送消息。针对延迟问题，使用异步发送方式会极大地提高系统的性能。

ActiveMQ缺省使用异步传输方式。但是按照JMS规范，当在事务外发送持久化消息的时候，ActiveMQ会强制使用同步发送方式。

在这种情况下，每一次发送都是同步的，而且阻塞到收到broker的应答。这个应答保证了broker已经成功地将消息持久化，

而且不会丢失。但是这样作也严重地影响了性能。

 

Message Cursors:

当producer发送的持久化消息到达broker之后，broker首先会把它保存在持久存储中。接下来，如果发现当前有活跃的consumer，如果这个consumer消费消息的速度能跟上producer生产消息的速度，那么ActiveMQ会直接把消息传递给broker内部跟这个consumer关联的dispatch queue；如果当前没有活跃的consumer或者consumer消费消息的速度跟不上producer生产消息的速度，那么ActiveMQ会使用Pending Message Cursors保存对消息的引用。

在需要的时候，Pending Message Cursors把消息引用传递给broker内部跟这个consumer关联的dispatch queue。以下是两种Pending Message Cursors：

VM Cursor:在内存中保存消息的引用。

File Cursor:首先在内存中保存消息的引用，如果内存使用量达到上限，那么会把消息引用保存到临时文件中。

在缺省情况下，ActiveMQ 会根据使用的Message Store来决定使用何种类型的Message Cursors，但是你可以根据destination来配置Message Cursors。

对于topic，可以使用的pendingSubscriberPolicy 有vmCursor和fileCursor。可以使用的PendingDurableSubscriberMessageStoragePolicy有

vmDurableCursor 和 fileDurableSubscriberCursor；对于queue，可以使用的pendingQueuePolicy有vmQueueCursor 和 fileQueueCursor。

 

Broker clusters:

The most common mental model of clustering in a JMS context is that there is a collection of JMS brokers and a JMS client will connect to one of them; then if the JMS broker goes down, it will auto-reconnect to another broker.We implement this using the failover:// protocol in the JMS client.

 

Master Slave:

The problem with running lots of stand alone brokers or brokers in a network is that messages are owned by a single physical broker at any point in time. If that broker goes down, you have to wait for it to be restarted before the message can be delivered. (If you are using non-persistent messaging and a broker goes down you generally lose your message).

The idea behind MasterSlave is that messaages are replicated to a slave broker so that even if you have catastrophic hardware failure of the master's machine, file system or data centre, you get immediate failover to the slave with no message loss.

Master Slave是目前ActiveMQ推荐的高可靠性和容错的解决方案。以下是几种不同的类型：

Pure Master Slave:

failover://(tcp://masterhost:61616,tcp://slavehost:61616)?randomize=false

设置randomize为false就可以让客户总是首先尝试连接master broker（slave broker并不会接受任何连接，直到它成为了master broker）。

工作方式：Master broker只有在消息成功被复制到slave broker之后才会响应客户；

Pure Master Slave具有以下限制：

只能有一个slave broker连接到master broker。

在因master broker失效而导致slave broker成为master之后，之前的master broker只有在当前的master broker（原slave broker）停止后才能重新生效。

Master broker失效后而切换到slave broker后，最安全的恢复master broker的方式是人工处理。首先要停止slave broker（这意味着所有的客户也要停止）。然后把slave broker的数据目录中所有的数据拷贝到master broker的数据目录中。然后重启master broker和slave broker。

Shared File System Master Slave

你可以运行多个broker，这些broker共享数据目录。当第一个broker得到文件上的排他锁之后，其它的broker便会在循环中等待获得这把锁。客户端使用failover transport来连接到可用的broker。当master broker失效的时候会释放这把锁，这时候其中一个slave broker会得到这把锁从而成为master broker。

JDBC Master Slave

JDBC Master Slave的工作原理跟Shared File System Master Slave类似，只是采用了数据库作为持久化存储

 

Replicated Message Stores:

An alternative to Master Slave is to have some way to replicate the message store; so for the disk files to be shared in some way. For example using a SAN or shared network drive you can share the files of a broker so that if it fails another broker can take over straight away. So by supporting a Replicated Message Store you can reduce the risk of message loss to provide either a HA backup or a full DR solution capable of surviving data centre failure.