Java中间件JMS(一)之ActiveMQ入门
1.ActiveMQ概述
企业消息软件从80年代起就存在,它不只是一种应用间消息传递风格,也是一种集成风格。因此,消息传递可以满足应用间的通知和互相操作。但是开源的解决方案是到最近10年才出现的。ApacheActiveMQ就是其中一种。它使应用间能以异步,松耦合方式交流。ActiveMQ是Apache出品,最流行的,能力强劲的开源消息总线。
ActiveMQ是Apache软件基金下的一个开源软件,它遵循JMS规范(JavaMessageService),是消息驱动中间件软件(MOM)。它为企业消息传递提供高可用,出色性能,可扩展,稳定和安全保障。ActiveMQ使用Apache许可协议。因此,任何人都可以使用和修改它而不必反馈任何改变。这对于商业上将ActiveMQ用在重要用途的人尤为关键。MOM的工作是在分布式的各应用之间调度事件和消息,使之到达指定的接收者。所以高可用,高性能,高可扩展性尤为关键。
2.ActiveMQ特性
⒈支持多种语言客户端,如:Java,C,C++,C#,Ruby,Perl,Python,PHP。应用协议有OpenWire,StompREST,WSNotification,XMPP,AMQP。
⒉完全支持JMS1.1和J2EE1.4规范,它们包括同步和异步消息传递,一次和只有一次的消息传递,对于预订者的持久消息等等。依附于JMS规范意味着,不论JMS消息提供者是谁,同样的基本特性(持久化,XA消息,事务)都是有效的。
⒊对Spring的支持,ActiveMQ可以很容易内嵌到使用Spring的系统里面去。
⒋通过了常见J2EE服务器(如Geronimo,JBoss4,GlassFish,WebLogic)的测试,其中通过JCA1.5resourceadaptors的配置,可以让ActiveMQ可以自动的部署到任何兼容J2EE1.4商业服务器上。
⒌ActiveMQ提供各种连接选择,包括HTTP,HTTPS,IP多点传送,SSL,STOMP,TCP,UDP,XMPP等。大量的连接协议支持使之具有更好的灵活性。很多现有的系统使用一种特定协议并且不能改变,所以一个支持多种协议的消息平台降低了使用的门槛。虽然连接很重要,但是和其他容器集成也同样重要。
6.ActiveMQ提供多种持久性方案可供选择,也可以完全按自己需求定制验证和授权。例如,ActiveMQ通过KahaDB提供自己的超快速消息持久方案(ultra-fastmessagepersistence),但也支持标准的JDBC方案。ActiveMQ可以通过配置文件提供简单的验证和授权,也提供标准的JAAS登陆模块。
7.ActiveMQ是为开发者设计的。它并不需要专门的管理工具,因为它提供各种易用且强大的管理特性。有很多方法去监控ActiveMQ的各个方面,可以通过JMX使用JConsole或ActiveMQwebconsole;可以运行ActiveMQ消息报告;可以用命令行脚本;可以通过日志。
8.代理器集群(Brokerclustering)----为了利于扩展,多个ActiveMQbroker能够联合工作。这个方式就是networkofbrokers并且能支持多种拓扑结构;支持客户端-服务器,点对点。
9.支持Ajax,支持与Axis的整合
3.ActiveMQ优势
1.与OpenJMS、JbossMQ等开源jmsprovider相比,ActiveMQ有Apache的支持,持续发展的优势明显。
2.速度很快,JBossMQ的十倍(没有测试)
3.提高系统资源的利用率,主要是任务的派发不是24小时平均的,而是高峰时期任务量很多,比如1秒1000多个,有的时候很低,比如十几秒钟才来一个。应用服务通过JMS队列一个一个的取任务,做完一个再领一个,使系统资源的运用趋于平均。而JMS,比如JMS接收消息的效率是很高的,比如ActiveMQ,在赛扬(2.40GHz)机器上能够达到2000/s,消息大小为1-2k。好一些的服务器可以达到2万以上/秒。
4.ActiveMQ安装
首先去http://activemq.apache.org/download.html下载最新版本ActiveMQ5.8.0Release,解压apache-activemq-5.8.0-bin.zip(或者apache-activemq-5.8.0-bin.tar.gz)
目录如下:
+bin(windows下面的bat和unix/linux下面的sh)
+conf(activeMQ配置目录,包含最基本的activeMQ配置文件)
+data(默认是空的)
+docs(只有index.html)
+example(几个例子)
+lib(activemMQ使用到的lib)
+webapps(后台管理页面)
+webapps-demo(后台管理消息发送页面)
+activemq-all-5.8.0.jar(java开发的jar包)
-LICENSE.txt
-NOTICE.txt
-README.txt
-user-guide.html
你可以使用bin\activemq.bat(activemq)启动。
启动ActiveMQ以后,登陆:http://localhost:8161/admin/默认用户/密码为admin/admin。
注意:
⒈这个仅仅是最基础的ActiveMQ的配置,很多地方都没有配置因此不要直接使用这个配置用于生存环境。
⒉有的时候由于端口被占用,导致ActiveMQ错误,ActiveMQ可能需要以下端口1099(JMX),61616(默认的TransportConnector)。
⒊如果没有物理网卡,或者MS的LoopBackAdpaterMulticast会报一个错误
5.运行附带的示例程序
1、Queue消息示例:
*启动Queue消息消费者
cdexample
antconsumer
*启动Queue消息生产者
cdexample
antproducer
简要说明:生产者(producer)发消息,消费者(consumer)接消息,发送/接收2000个消息后自动关闭
2、Topic消息示例:
*启动Topic消息消费者
cdexample
anttopic-listener
*启动Topic消息生产者
cdexample
anttopic-publisher
简要说明:重复10轮,publisher每轮发送2000个消息,并等待获取listener的处理结果报告,然后进入下一轮发送,最后
统计全局发送时间。
3、Queue消息和Topic消息发送之后,可以通过后台点击Queues和Topics查看发送消息具体信息。
6.ActiveMQ类别及开发流程
1)、Point-to-Point(点对点)消息模式开发流程:
1、生产者(producer)开发流程(ProducerTool.java):
1.1创建Connection:根据url,user和password创建一个jmsConnection。
1.2创建Session:在connection的基础上创建一个session,同时设置是否支持事务和ACKNOWLEDGE标识。
1.3创建Destination对象:需指定其对应的主题(subject)名称,producer和consumer将根据subject来发送/接收对应的消息。
1.4创建MessageProducer:根据Destination创建MessageProducer对象,同时设置其持久模式。
1.5发送消息到队列(Queue):封装TextMessage消息,使用MessageProducer的send方法将消息发送出去。
2、消费者(consumer)开发流程(ConsumerTool.java):
2.1实现MessageListener接口:消费者类必须实现MessageListener接口,然后在onMessage()方法中监听消息的到达并处理。
2.2创建Connection:根据url,user和password创建一个jmsConnection,如果是durable模式,还需要给connection设置一个clientId。
2.3创建Session和Destination:与ProducerTool.java中的流程类似,不再赘述。
2.4创建replyProducer【可选】:可以用来将消息处理结果发送给producer。
2.5创建MessageConsumer:根据Destination创建MessageConsumer对象。
2.6消费message:在onMessage()方法中接收producer发送过来的消息进行处理,并可以通过replyProducer反馈信息给producer
[java]viewplaincopy
if(message.getJMSReplyTo()!=null){
replyProducer.send(message.getJMSReplyTo(),
session.createTextMessage("Reply:"+message.getJMSMessageID()));
2)、Publisher/Subscriber(发布/订阅者)消息模式开发流程
1、订阅者(Subscriber)开发流程(TopicListener.java):
1.1实现MessageListener接口:在onMessage()方法中监听发布者发出的消息队列,并做相应处理。
1.2创建Connection:根据url,user和password创建一个jmsConnection。
1.3创建Session:在connection的基础上创建一个session,同时设置是否支持事务和ACKNOWLEDGE标识。
1.4创建Topic:创建2个Topic,topictest.messages用于接收发布者发出的消息,topictest.control用于向发布者发送消息,实现双方的交互。
1.5创建consumer和producer对象:根据topictest.messages创建consumer,根据topictest.control创建producer。
1.6接收处理消息:在onMessage()方法中,对收到的消息进行处理,可直接简单在本地显示消息,或者根据消息内容不同处理对应的业务逻辑(比如:数据库更新、文件操作等等),并且可以使用producer对象将处理结果返回给发布者。
2、发布者(Publisher)开发流程(TopicPublisher.java):
2.1实现MessageListener接口:在onMessage()方法中接收订阅者的反馈消息。
2.2创建Connection:根据url,user和password创建一个jmsConnection。
2.3创建Session:在connection的基础上创建一个session,同时设置是否支持事务和ACKNOWLEDGE标识。
2.4创建Topic:创建2个Topic,topictest.messages用于向订阅者发布消息,topictest.control用于接收订阅者反馈的消息。这2个topic与订阅者开发流程中的topic是一一对应的。
2.5创建consumer和producer对象:根据topictest.messages创建publisher;根据topictest.control创建consumer,同时监听订阅者反馈的消息。
2.6给所有订阅者发送消息,并接收反馈消息:示例代码中,一共重复10轮操作。每轮先向所有订阅者发送2000个消息;然后堵塞线程,开始等待;最后通过onMessage()方法,接收到订阅者反馈的“REPORT”类信息后,才print反馈信息并解除线程堵塞,进入下一轮。
注:可同时运行多个订阅者测试查看此模式效果
7.Eclipse代码开发
1.建立一个WebProbject项目,将activemq-all-5.8.0.jar放在lib里面
2.Queue(点对点)方式:生产者
package jms; import javax.jms.Connection; import javax.jms.ConnectionFactory; import javax.jms.DeliveryMode; import javax.jms.Destination; import javax.jms.MessageProducer; import javax.jms.Queue; import javax.jms.Session; import javax.jms.TextMessage; import org.apache.activemq.ActiveMQConnection; import org.apache.activemq.ActiveMQConnectionFactory; //Queue(点对点)方式 生存者Producer public class QueueProducer { private static String user = ActiveMQConnection.DEFAULT_USER; private static String password =ActiveMQConnection.DEFAULT_PASSWORD; private static String url = "tcp://localhost:61616"; public static void main(String[] args)throws Exception { // ConnectionFactory :连接工厂,JMS 用它创建连接 ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(user,password,url); // Connection :JMS 客户端到JMS Provider 的连接 Connection connection = connectionFactory.createConnection(); // Connection 启动 connection.start(); System.out.println("Connection is start..."); // Session: 一个发送或接收消息的线程 Session session = connection.createSession(Boolean.TRUE,Session.AUTO_ACKNOWLEDGE); // Queue :消息的目的地;消息发送给谁. Queue destination = session.createQueue("example.A"); // MessageProducer:消息发送者 MessageProducer producer = session.createProducer(destination); // 设置不持久化,此处学习,实际根据项目决定 producer.setDeliveryMode(DeliveryMode.PERSISTENT); // 构造消息,此处写死,项目就是参数,或者方法获取 sendMessage(session, producer); session.commit(); connection.close(); System.out.println("send text ok."); } public static void sendMessage(Session session, MessageProducer producer) throws Exception { for (int i = 1; i <= 100; i++) {//有限制,达到1000就不行 TextMessage message = session.createTextMessage("ActiveMq 发送的消息" + i); // 发送消息到目的地方 System.out.println("发送消息:" + "ActiveMq 发送的消息" + i); producer.send(message); } } }
3.Queue(点对点)方式:消费者
package jms; import javax.jms.Connection; import javax.jms.ConnectionFactory; import javax.jms.Destination; import javax.jms.JMSException; import javax.jms.Message; import javax.jms.MessageConsumer; import javax.jms.MessageListener; import javax.jms.Queue; import javax.jms.Session; import javax.jms.TextMessage; import org.apache.activemq.ActiveMQConnection; import org.apache.activemq.ActiveMQConnectionFactory; //Queue(点对点)方式 消费这Consumer public class QueueConsumer { private static String user = ActiveMQConnection.DEFAULT_USER; private static String password =ActiveMQConnection.DEFAULT_PASSWORD; private static String url = "tcp://localhost:61616"; public static void main(String[] args) throws Exception{ // ConnectionFactory :连接工厂,JMS 用它创建连接 ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(user,password,url); // Connection :JMS 客户端到JMS Provider 的连接 Connection connection = connectionFactory.createConnection(); connection.start(); // Session: 一个发送或接收消息的线程 final Session session = connection.createSession(Boolean.TRUE, Session.AUTO_ACKNOWLEDGE); // Destination :消息的目的地;消息发送给谁. Queue destination=session.createQueue("example.A"); // 消费者,消息接收者 MessageConsumer consumer = session.createConsumer(destination); consumer.setMessageListener(new MessageListener(){//有事务限制 @Override public void onMessage(Message message) { try { TextMessage textMessage=(TextMessage)message; System.out.println(textMessage.getText()); } catch (JMSException e1) { e1.printStackTrace(); } try { session.commit(); } catch (JMSException e) { e.printStackTrace(); } } }); /* 另外一种接受方式 * while (true) { //设置接收者接收消息的时间,为了便于测试,这里谁定为100s TextMessage message = (TextMessage) consumer.receive(100000); if (null != message) { System.out.println("收到消息" + message.getText()); } else { break; } }*/ } }
4.Topic(发布/订阅)方式:发布者
package jms; import java.util.Date; import javax.jms.Connection; import javax.jms.ConnectionFactory; import javax.jms.DeliveryMode; import javax.jms.JMSException; import javax.jms.MapMessage; import javax.jms.MessageProducer; import javax.jms.Queue; import javax.jms.Session; import javax.jms.TextMessage; import javax.jms.Topic; import org.apache.activemq.ActiveMQConnection; import org.apache.activemq.ActiveMQConnectionFactory; //Topic(发布/订阅)方式 发布者Publisher public class TopicPublisher { private static String user = ActiveMQConnection.DEFAULT_USER; private static String password =ActiveMQConnection.DEFAULT_PASSWORD; private static String url = "tcp://localhost:61616"; public static void main(String[] args)throws Exception { // ConnectionFactory :连接工厂,JMS 用它创建连接 ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(user,password,url); // Connection :JMS 客户端到JMS Provider 的连接 Connection connection = connectionFactory.createConnection(); // Connection 启动 connection.start(); System.out.println("Connection is start..."); // Session: 一个发送或接收消息的线程 Session session = connection.createSession(Boolean.TRUE,Session.AUTO_ACKNOWLEDGE); // Topicr :消息的目的地;消息发送给谁. Topic destination = session.createTopic("example.A"); // MessageProducer:消息发送者 MessageProducer producer = session.createProducer(destination); // 设置不持久化,此处学习,实际根据项目决定 producer.setDeliveryMode(DeliveryMode.PERSISTENT); // 构造消息,此处写死,项目就是参数,或者方法获取 sendMessage(session, producer); session.commit(); connection.close(); System.out.println("send text ok."); } public static void sendMessage(Session session, MessageProducer producer) throws Exception { for (int i = 1; i <= 100; i++) {//有限制,达到1000就不行 TextMessage message = session.createTextMessage("ActiveMq 发送的消息" + i); // 发送消息到目的地方 System.out.println("发送消息:" + "ActiveMq 发送的消息" + i); producer.send(message); } } }
5.Topic(发布/订阅)方式:订阅者
package jms; import javax.jms.Connection; import javax.jms.ConnectionFactory; import javax.jms.JMSException; import javax.jms.Message; import javax.jms.MessageConsumer; import javax.jms.MessageListener; import javax.jms.Session; import javax.jms.TextMessage; import javax.jms.Topic; import org.apache.activemq.ActiveMQConnection; import org.apache.activemq.ActiveMQConnectionFactory; //Topic(发布/订阅)方式 订阅者TopicSubscriber public class TopicSubscriber { private static String user = ActiveMQConnection.DEFAULT_USER; private static String password =ActiveMQConnection.DEFAULT_PASSWORD; private static String url = "tcp://localhost:61616"; public static void main(String[] args) throws Exception{ // ConnectionFactory :连接工厂,JMS 用它创建连接 ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(user,password,url); // Connection :JMS 客户端到JMS Provider 的连接 Connection connection = connectionFactory.createConnection(); connection.start(); // Session: 一个发送或接收消息的线程 final Session session = connection.createSession(Boolean.TRUE, Session.AUTO_ACKNOWLEDGE); // Destination :消息的目的地;消息发送给谁. Topic destination=session.createTopic("example.A"); // 消费者,消息接收者 MessageConsumer consumer = session.createConsumer(destination); consumer.setMessageListener(new MessageListener(){//有事务限制 @Override public void onMessage(Message message) { try { TextMessage textMessage=(TextMessage)message; System.out.println(textMessage.getText()); } catch (JMSException e1) { e1.printStackTrace(); } try { session.commit(); } catch (JMSException e) { e.printStackTrace(); } } }); /* 另外一种接受方式 * while (true) { //设置接收者接收消息的时间,为了便于测试,这里谁定为100s TextMessage message = (TextMessage) consumer.receive(100000); if (null != message) { System.out.println("收到消息" + message.getText()); } else { break; } }*/ } }
Queue(点对点)方式和Topic(发布/订阅)方式的运行结果最明显的区别是:在Queue(点对点)方式中先运行生产者,再运行消费者,消费者还能接受到消息;
而Topic(发布/订阅)方式就不同了,先运行发布者,再运行订阅者,订阅者收到的消息
可能没有或者是不完全的。