JMS-ActiveMQ入门实例

 

下载ActiveMQ http://activemq.apache.org/download.html

解压缩到本地

 

启动mq：/bin/activemq.bat

 

管理界面： http://localhost:8161/admin，默认不用验证。

如果加验证可以参考http://wjw465150.iteye.com/blog/479527

 

发送消息测试：

package test;

import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.DeliveryMode;
import javax.jms.Destination;
import javax.jms.MessageProducer;
import javax.jms.Session;
import javax.jms.TextMessage;
import org.apache.activemq.ActiveMQConnection;
import org.apache.activemq.ActiveMQConnectionFactory;

public class Sender {

	private static final int SEND_NUMBER = 5;

	public static void main(String[] args) {
		// ConnectionFactory ：连接工厂，JMS 用它创建连接
		ConnectionFactory connectionFactory;
		// Connection ：JMS 客户端到JMS Provider 的连接
		Connection connection = null;
		// Session： 一个发送或接收消息的线程
		Session session;
		// Destination ：消息的目的地;消息发送给谁.
		Destination destination;
		// MessageProducer：消息发送者
		MessageProducer producer;
		// TextMessage message;
		// 构造ConnectionFactory实例对象，此处采用ActiveMq的实现jar
		connectionFactory = new ActiveMQConnectionFactory(
				ActiveMQConnection.DEFAULT_USER,
				ActiveMQConnection.DEFAULT_PASSWORD, "tcp://localhost:61616");

		try {
			// 构造从工厂得到连接对象
			connection = connectionFactory.createConnection();
			// 启动
			connection.start();
			// 获取操作连接
			session = connection.createSession(Boolean.TRUE.booleanValue(),
					Session.AUTO_ACKNOWLEDGE);
			// 获取session注意参数值xingbo.xu-queue是一个服务器的queue，须在在ActiveMq的console配置
			destination = session.createQueue("xingbo.xu-queue");
			// 得到消息生成者【发送者】
			producer = session.createProducer(destination);
			// 设置不持久化，此处学习，实际根据项目决定
			producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);
			// 构造消息，此处写死，项目就是参数，或者方法获取
			sendMessage(session, producer);
			session.commit();
		} catch (Exception e) {
			e.printStackTrace();
		} finally {
			try {
				if (null != connection)
					connection.close();
			} catch (Throwable ignore) {
			}
		}
	}

	public static void sendMessage(Session session, MessageProducer producer)
			throws Exception {
		for (int i = 1; i <= SEND_NUMBER; i++) {
			TextMessage message = session
					.createTextMessage("青软学生学习ActiveMq 发送的消息" + i);
			// 发送消息到目的地方
			System.out.println("发送消息：" + "青软学生学习ActiveMq 发送的消息" + i);
			producer.send(message);
		}
	}
}

 

接受代码：

package test;

import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.Destination;
import javax.jms.MessageConsumer;
import javax.jms.Session;
import javax.jms.TextMessage;
import org.apache.activemq.ActiveMQConnection;
import org.apache.activemq.ActiveMQConnectionFactory;

public class Receiver {
	public static void main(String[] args) {
		// ConnectionFactory ：连接工厂，JMS 用它创建连接
		ConnectionFactory connectionFactory;
		// Connection ：JMS 客户端到JMS Provider 的连接
		Connection connection = null;
		// Session： 一个发送或接收消息的线程
		Session session;
		// Destination ：消息的目的地;消息发送给谁.
		Destination destination;
		// 消费者，消息接收者
		MessageConsumer consumer;
		connectionFactory = new ActiveMQConnectionFactory(
				ActiveMQConnection.DEFAULT_USER,
				ActiveMQConnection.DEFAULT_PASSWORD, "tcp://localhost:61616");

		try {
			// 构造从工厂得到连接对象
			connection = connectionFactory.createConnection();
			// 启动
			connection.start();
			// 获取操作连接
			session = connection.createSession(Boolean.FALSE.booleanValue(),
					Session.AUTO_ACKNOWLEDGE);
			// 获取session注意参数值xingbo.xu-queue是一个服务器的queue，须在在ActiveMq的console配置
			destination = session.createQueue("xingbo.xu-queue");
			consumer = session.createConsumer(destination);
			while (true) {
				TextMessage message = (TextMessage) consumer.receive(1000);
				if (null != message) {
					System.out.println("收到消息" + message.getText());
				} else {
					break;
				}
			}
		} catch (Exception e) {
			e.printStackTrace();
		} finally {
			try {
				if (null != connection)
					connection.close();
			} catch (Throwable ignore) {

			}
		}
	}
}

 

依赖的2个包：

activemq-core-5.3.0.jar

commons-logging-1.1.jar

jdk必须使用1.5+

 

=======附加资料===========================

关于ActiveMQ

ActiveMQ 是Apache出品，最流行的，能力强劲的开源消息总线。ActiveMQ 是一个完全支持JMS1.1和J2EE 1.4规范的 JMS Provider实现,尽管JMS规范出台已经是很久的事情了,但是JMS在当今的J2EE应用中间仍然扮演着特殊的地位。

 

ActiveMQ特性列表

1. 多种语言和协议编写客户端。语言: Java, C, C++, C#, Ruby, Perl, Python, PHP。应用协议: OpenWire,Stomp REST,WS Notification,XMPP,AMQP
　　2. 完全支持JMS1.1和J2EE 1.4规范 (持久化,XA消息,事务)
　　3. 对Spring的支持,ActiveMQ可以很容易内嵌到使用Spring的系统里面去,而且也支持Spring2.0的特性
　　4. 通过了常见J2EE服务器(如 Geronimo,JBoss 4, GlassFish,WebLogic)的测试,其中通过JCA 1.5 resource adaptors的配置,可以让ActiveMQ可以自动的部署到任何兼容J2EE 1.4 商业服务器上
　　5. 支持多种传送协议:in-VM,TCP,SSL,NIO,UDP,JGroups,JXTA
　　6. 支持通过JDBC和journal提供高速的消息持久化
　　7. 从设计上保证了高性能的集群,客户端-服务器,点对点
　　8. 支持Ajax
　　9. 支持与Axis的整合
　　10. 可以很容易得调用内嵌JMS provider,进行测试

 

 

连接方式：
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.

 

====================================================

推荐两个简单消息服务实现

1、wjw纯java写的简单消息队列服务sqs4j,每秒可以并发8000/s,地址http://sourceforge.net/projects/sqs4j/

2、张宴的用纯C写的一个开源项目HTTPSQS,并发可以达到15000/s，地址 http://blog.s135.com/httpsqs/

两者协议一模一样,只是sqs4j用纯java实现的,网络层用的Netty,数据库层用的Berkeley

 

 

 

评论

1 楼 luckywnj 2013-02-01  

3q楼主，受益匪浅！

想请教个问题，小弟用
2.Embedded Broker
以编码的方式启动broker:
方式启动

没有用spring，也没有用配置文件，可以发送也可以接受，
但是不能管理
http://localhost:8161/admin/
这个URL打不开，8161端口也没打开

请问怎么才能打开这个URL？