【一头扎进JMS】(2)----ActiviteMQ点对点消息实现

待我称王封你为后i 2022-06-14 04:48 130阅读 0赞

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

**特性:**

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

**安装:**

**主页:**

**[http://activemq.apache.org/][http_activemq.apache.org]**

**下载activemq的安装包.我下载的是windows的apache-activemq-5.14.0版本.**

**activemq服务启动地址:http://127.0.0.1:8161/admin/. 用户名/密码:admin/admin**

**启动方式:在apache-activemq-5.14.0的bin目录下有win32和win64文件夹,我的电脑是64位的,启动的时候选择win64文件夹下的activemq.bat执行.不要直接执行apache-activemq-5.14.0/bin目录下的activemq.bat.**

**![Center][]**

**![Center 1][]**

**可视化界面:**

**![Center 2][]**

**实例讲解:**

**消息生产者:**

package com.java1234.activemq; import javax.jms.Connection; import javax.jms.ConnectionFactory; import javax.jms.Destination; import javax.jms.JMSException; import javax.jms.MessageProducer; import javax.jms.Session; import javax.jms.TextMessage; import org.apache.activemq.ActiveMQConnection; import org.apache.activemq.ActiveMQConnectionFactory; /** * 消息生产者 * @author Administrator * */ public class JMSProducer { private static final String USERNAME=ActiveMQConnection.DEFAULT_USER; // 默认的连接用户名 private static final String PASSWORD=ActiveMQConnection.DEFAULT_PASSWORD; // 默认的连接密码 private static final String BROKEURL=ActiveMQConnection.DEFAULT_BROKER_URL; // 默认的连接地址 private static final int SENDNUM=10; // 发送的消息数量 public static void main(String[] args) { ConnectionFactory connectionFactory; // 连接工厂 Connection connection = null; // 连接 Session session; // 会话 接受或者发送消息的线程 Destination destination; // 消息的目的地 MessageProducer messageProducer; // 消息生产者 // 实例化连接工厂 connectionFactory=new ActiveMQConnectionFactory(JMSProducer.USERNAME, JMSProducer.PASSWORD, JMSProducer.BROKEURL); try { connection=connectionFactory.createConnection(); // 通过连接工厂获取连接 connection.start(); // 启动连接 session=connection.createSession(Boolean.TRUE, Session.AUTO_ACKNOWLEDGE); // 创建Session destination=session.createQueue("FirstQueue1"); // 创建消息队列 messageProducer=session.createProducer(destination); // 创建消息生产者 sendMessage(session, messageProducer); // 发送消息 session.commit(); } catch (Exception e) { // TODO Auto-generated catch block e.printStackTrace(); } finally{ if(connection!=null){ try { connection.close(); } catch (JMSException e) { // TODO Auto-generated catch block e.printStackTrace(); } } } } /** * 发送消息 * @param session * @param messageProducer * @throws Exception */ public static void sendMessage(Session session,MessageProducer messageProducer)throws Exception{ for(int i=0;i<JMSProducer.SENDNUM;i++){ TextMessage message=session.createTextMessage("ActiveMQ 发送的消息"+i); System.out.println("发送消息："+"ActiveMQ 发送的消息"+i); messageProducer.send(message); } } }

**执行结果:**

**生产者发送消息:**

**![Center 3][]**

**![Center 4][]**

**消息消费者:**

package com.java1234.activemq; import javax.jms.Connection; import javax.jms.ConnectionFactory; import javax.jms.Destination; import javax.jms.JMSException; import javax.jms.MessageConsumer; import javax.jms.Session; import javax.jms.TextMessage; import org.apache.activemq.ActiveMQConnection; import org.apache.activemq.ActiveMQConnectionFactory; /** * 消息消费者 * @author Administrator * */ public class JMSConsumer { private static final String USERNAME=ActiveMQConnection.DEFAULT_USER; // 默认的连接用户名 private static final String PASSWORD=ActiveMQConnection.DEFAULT_PASSWORD; // 默认的连接密码 private static final String BROKEURL=ActiveMQConnection.DEFAULT_BROKER_URL; // 默认的连接地址 public static void main(String[] args) { ConnectionFactory connectionFactory; // 连接工厂 Connection connection = null; // 连接 Session session; // 会话 接受或者发送消息的线程 Destination destination; // 消息的目的地 MessageConsumer messageConsumer; // 消息的消费者 // 实例化连接工厂 connectionFactory=new ActiveMQConnectionFactory(JMSConsumer.USERNAME, JMSConsumer.PASSWORD, JMSConsumer.BROKEURL); try { connection=connectionFactory.createConnection(); // 通过连接工厂获取连接 connection.start(); // 启动连接 session=connection.createSession(Boolean.FALSE, Session.AUTO_ACKNOWLEDGE); // 创建Session destination=session.createQueue("FirstQueue1"); // 创建连接的消息队列 messageConsumer=session.createConsumer(destination); // 创建消息消费者 while(true){ TextMessage textMessage=(TextMessage)messageConsumer.receive(100000); if(textMessage!=null){ System.out.println("收到的消息："+textMessage.getText()); }else{ break; } } } catch (JMSException e) { // TODO Auto-generated catch block e.printStackTrace(); } } }

**执行结果:**

**消费者执行结果:**

**![Center 5][]**

**![Center 6][]**

[http_activemq.apache.org]: http://activemq.apache.org/
[Center]: /images/20220614/507ee68f412146a6ad85bda9dc1c3ae4.png
[Center 1]: /images/20220614/6c584db74edf4b199cea63ec7d28d9ec.png
[Center 2]: /images/20220614/341ff1a88d794fdeb0e8ea8baa07f3fa.png
[Center 3]: /images/20220614/a39b2c882708489090578d91622efca1.png
[Center 4]: /images/20220614/cb2164e663824ea5bbafef7ffb6daf09.png
[Center 5]: /images/20220614/e0d9527d772642feb8491b70ad121c7f.png
[Center 6]: /images/20220614/d099aaace1604f01af55a3c47d0950af.png