Java中间件JMS之ActiveMQ入门-低调大师

Java中间件JMS之ActiveMQ入门

2017-11-14 449

文章原文:http://blog.csdn.net/dengwanchuan/article/details/10241345

1.ActiveMQ概述

企业消息软件从80年代起就存在，它不只是一种应用间消息传递风格，也是一种集成风格。因此，消息传递可以满足应用间的通知和互相操作。但是开源的解决方案是到最近10年才出现的。Apache ActiveMQ就是其中一种。它使应用间能以异步，松耦合方式交流。ActiveMQ 是Apache出品，最流行的，能力强劲的开源消息总线。

‍ ActiveMQ是Apache软件基金下的一个开源软件，它遵循JMS规范（Java Message Service），是消息驱动中间件软件（MOM）。它为企业消息传递提供高可用，出色性能，可扩展，稳定和安全保障。ActiveMQ使用Apache许可协议。因此，任何人都可以使用和修改它而不必反馈任何改变。这对于商业上将ActiveMQ用在重要用途的人尤为关键。MOM的工作是在分布式的各应用之间调度事件和消息，使之到达指定的接收者。所以高可用，高性能，高可扩展性尤为关键。

2.ActiveMQ特性

⒈支持多种语言客户端,如:Java,C,C++,C#,Ruby,Perl,Python,PHP。应用协议有 OpenWire,Stomp REST,WS Notification,XMPP,AMQP。

⒉ 完全支持JMS1.1和J2EE1.4规范,它们包括同步和异步消息传递，一次和只有一次的消息传递，对于预订者的持久消息等等。依附于JMS规范意味着，不论JMS消息提供者是谁，同样的基本特性（持久化，XA消息，事务)都是有效的。

⒊ 对Spring的支持，ActiveMQ可以很容易内嵌到使用Spring的系统里面去。

⒋ 通过了常见J2EE服务器（如 Geronimo,JBoss 4,GlassFish,WebLogic)的测试，其中通过JCA 1.5 resource adaptors的配置，可以让ActiveMQ可以自动的部署到任何兼容J2EE 1.4 商业服务器上。

⒌ ActiveMQ提供各种连接选择，包括HTTP，HTTPS，IP多点传送，SSL，STOMP，TCP，UDP，XMPP等。大量的连接协议支持使之具有更好的灵活性。很多现有的系统使用一种特定协议并且不能改变，所以一个支持多种协议的消息平台降低了使用的门槛。虽然连接很重要，但是和其他容器集成也同样重要。

6.ActiveMQ提供多种持久性方案可供选择，也可以完全按自己需求定制验证和授权。例如，ActiveMQ通过KahaDB提供自己的超快速消息持久方案（ultra-fast message persistence），但也支持标准的JDBC方案。ActiveMQ可以通过配置文件提供简单的验证和授权，也提供标准的JAAS登陆模块。

7.ActiveMQ是为开发者设计的。它并不需要专门的管理工具，因为它提供各种易用且强大的管理特性。有很多方法去监控ActiveMQ的各个方面，可以通过JMX使用JConsole或ActiveMQ web console；可以运行ActiveMQ消息报告；可以用命令行脚本；可以通过日志。

8.代理器集群（Broker clustering）----为了利于扩展，多个ActiveMQ broker能够联合工作。这个方式就是network of brokers并且能支持多种拓扑结构;支持客户端-服务器，点对点。

9.支持Ajax, 支持与Axis的整合

3.ActiveMQ优势

1.与OpenJMS、JbossMQ等开源jms provider相比，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 下载最新版本ActiveMQ 5.8.0 Release，解压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的LoopBackAdpater Multicast会报一个错误

5.运行附带的示例程序

1、Queue消息示例：

* 启动Queue消息消费者

cd example

ant consumer

* 启动Queue消息生产者

cd example

ant producer
简要说明：生产者（producer）发消息，消费者（consumer）接消息，发送/接收2000个消息后自动关闭

2、Topic消息示例：

* 启动Topic消息消费者

cd example

ant topic-listener

* 启动Topic消息生产者

cd example

ant topic-publisher
简要说明：重复10轮，publisher每轮发送2000个消息，并等待获取listener的处理结果报告，然后进入下一轮发送，最后

统计全局发送时间。

3、Queue消息和Topic消息发送之后，可以通过后台点击Queues和Topics查看发送消息具体信息。

SouthEast

6.ActiveMQ类别及开发流程

1)、Point-to-Point (点对点)消息模式开发流程：
1、生产者（producer）开发流程（ProducerTool.java）：

1.1 创建Connection：根据url，user和password创建一个jms Connection。

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创建一个jms Connection，如果是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]view plaincopy
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创建一个jms Connection。

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创建一个jms Connection。

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.建立一个Web Probject 项目,将activemq-all-5.8.0.jar放在lib里面

2.Queue（点对点）方式：生产者

[java]view plaincopy
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
publicclass QueueProducer {  
privatestatic String user = ActiveMQConnection.DEFAULT_USER;  
privatestatic String password =ActiveMQConnection.DEFAULT_PASSWORD;  
privatestatic String url =  "tcp://localhost:61616";  
publicstaticvoid 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.");  
   }  
publicstaticvoid 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（点对点）方式：消费者

[java]view plaincopy
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
publicclass QueueConsumer {  
privatestatic String user = ActiveMQConnection.DEFAULT_USER;  
privatestatic String password =ActiveMQConnection.DEFAULT_PASSWORD;  
privatestatic String url = "tcp://localhost:61616";  
publicstaticvoid 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
publicvoid 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（发布/订阅）方式：发布者

[java]view plaincopy
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
publicclass TopicPublisher {  
privatestatic String user = ActiveMQConnection.DEFAULT_USER;  
privatestatic String password =ActiveMQConnection.DEFAULT_PASSWORD;  
privatestatic String url =  "tcp://localhost:61616";  
publicstaticvoid 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.");  
   }  
publicstaticvoid 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（发布/订阅）方式：订阅者

[java]view plaincopy
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
publicclass TopicSubscriber {  
privatestatic String user = ActiveMQConnection.DEFAULT_USER;  
privatestatic String password =ActiveMQConnection.DEFAULT_PASSWORD;  
privatestatic String url = "tcp://localhost:61616";  
publicstaticvoid 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
publicvoid 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（发布/订阅）方式就不同了，先运行发布者，再运行订阅者，订阅者收到的消息

可能没有或者是不完全的。

本文转自yunlielai51CTO博客，原文链接： http://blog.51cto.com/4925054/1283320 ，如需转载请自行联系原作者

微信关注我们

原文链接：https://yq.aliyun.com/articles/550442

转载内容版权归作者及来源网站所有！

低调大师中文资讯倾力打造互联网数据资讯、行业资源、电子商务、移动互联网、网络营销平台。持续更新报道IT业界、互联网、市场资讯、驱动更新,是最及时权威的产业资讯及硬件资讯报道平台。

我的2012

以往都是写公司的年终总结，今年新入职的公司少了这一环，干脆就把年终总结写在博客上，给自己这一年的工作和生活留下点回忆，总结下得失。离职: 前半年，继续在曾经的公司延续着枯燥的工作和生活，现在这家公司再也没有三年前我入职时感觉到的那种亲近的感觉，也再也没有半点“研发”的感觉，也许是公司大了，官僚主义在蔓延；也许是自己在方方面面都受到了冷落；也许是公司领导曾说的一句话深深地刺痛了我做研发人员的心：“我们有着三流的研发，一流的市场。”（在公司领导的眼里，我们的市场都是销售人员喝酒喝出来的，不是靠技术争取来的，其实也难怪，我们的客户都是些容忍度极高的公务员和事业人员）；也许是曾经的那一票兄弟基本上都走了，没走的也因为别的原因转到了别的岗位，让我倍感孤独；也许是我再也看不下去曾经大家认为技术最差的某个同事因为溜须拍马竟获得了晋升；但最终是因为我认为这家公司已经不值得我为他再蹉跎我的青春，消磨我的梦想，我必须离开了。新起点：来到一个新的公司，一个正在成长期的公司，进入了一个刚成立的部门——核心技术研究院，顶头上司就是这个公司的曾经研发总监，很荣幸我作为一个曾经有着六年工作经验的”...

2017-11-15

579

优点：配置简单更灵活高并发（静态小文件）占用资源少功能种类多支持epoll模型配合动态服务和apache有区别对IP限速，可以限制链接数 LNMP实现原理架构： nginx 80 fastcgi_pass http:ip:9000 PHP FCGI（服务） 9000端口 apache 是模块 DSO apxs select模型 nginx是起CGI epoll网络I/O模型 nginx.org # ntpdate cn.pool.ntp.org 更新系统时间开始安装 wgethttp://nginx.org/download/nginx-1.9.15.tar.gz 安装准备安装 pcre 中文perl兼容正则表达式为了支持http rewrite 伪静态安装 openssl yum -y install pcre* openssl openssl-devel useradd -s /sbin/nologin -M nginx wgethttp://nginx.org/downloa...

2017-11-15

568

资源下载

更多资源

优质分享App

近一个月的开发和优化，本站点的第一个app全新上线。该app采用极致压缩，本体才4.36MB。系统里面做了大量数据访问、缓存优化。方便用户在手机上查看文章。后续会推出HarmonyOS的适配版本。

Mario

马里奥是站在游戏界顶峰的超人气多面角色。马里奥靠吃蘑菇成长，特征是大鼻子、头戴帽子、身穿背带裤，还留着胡子。与他的双胞胎兄弟路易基一起，长年担任任天堂的招牌角色。

腾讯云软件源

为解决软件依赖安装时官方源访问速度慢的问题，腾讯云为一些软件搭建了缓存服务。您可以通过使用腾讯云软件源站来提升依赖包的安装速度。为了方便用户自由搭建服务架构，目前腾讯云软件源站支持公网访问和内网访问。

Spring

Spring框架（Spring Framework）是由Rod Johnson于2002年提出的开源Java企业级应用框架，旨在通过使用JavaBean替代传统EJB实现方式降低企业级编程开发的复杂性。该框架基于简单性、可测试性和松耦合性设计理念，提供核心容器、应用上下文、数据访问集成等模块，支持整合Hibernate、Struts等第三方框架，其适用范围不仅限于服务器端开发，绝大多数Java应用均可从中受益。