ActiveMQ实战

ActiveMQ是JMS规范的一个实现产品，即对JMS的一套接口进行了实现。所以下面我们先对JMS与ActiveMQ进行一个简单介绍，然后再介绍ActiveMQ的P2P与PubSub两种开发模式。

一、    JMS简介

       JMS即Java消息服务，是J2EE的13规范之一，是一套面向消息中间件（MOM）的应用程序接口。用在两个程序之间，或分布式系统中发送消息，进行异步通信。

       JMS使我们能够通过消息收发服务（也可以称为，消息中介程序或路由器）从一个JMS客户机向另一个JMS客户机发送消息。消息是JMS中的一种类型对象，由两部分组成：报头和消息主体。报头由路由信息以及有关该消息的元数据组成。消息主体则携带着应用程序的数据或有效负载。根据有效负载的类型来划分，消息可分为多种，它们分别携带：简单文本消息、可序列化的对象、属性集合、字节流、原始值流、无有效负载消息。

二、    ActiveMQ简介

       ActiveMQ出身名门，是Apache门下的最流行的，能力强劲的开源消息总线。完全支持JMS1.1和J2EE1.4规范的JMS Provide实现。它从设计上保证了高性能的集群，当然实现了JMS的P2P与PubSub两种开发模式。

三、    安装运行ActiveMQ

       首先下载ActiveMQ的稳定版：http://activemq.apache.org ，如下图，我下载的是apache-activemq-5.11.1-bin：

　　


       然后进入到bin/win64文件夹下，并启动activemq.bat：

　　


       出现如下提示表示ActiveMQ启动成功：

　　


       然后在浏览器输入：http://localhost:8161/即可看到ActiveMQ的管理界面：

　　


       然后我们点击：Manage ActiveMQ broker登录管理后台：（用户名、密码默认都为admin）

　　


       登录进去以后我们可以点击Queues看到消息队列以及消费者情况等：

　　


       ActiveMQ的环境已经好了，下面我们开始开发P2P与PubSub的实例。

四、    P2P开发模式实战

       建立JavaProject：ActiveMQ-p2p，并导入ActiveMQ的核心jar包，只需要导入根目录下的activemq-all.jar即可：

　　


       然后建立一个消息生产者类：JMSProducer；一个消息消费者类：JMSConsumer：

       源码如下：

       消息生产者类：JMSProducer

[java] view
plain copy

package com.tgb.jms.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) {  

            e.printStackTrace();  

        } finally {  

            if (connection != null) {  

                try {  

                    connection.close();  

                } catch (JMSException e) {  

                    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);  

        }  

    }  

}  

       消息消费者类：JMSConsumer

[java] view
plain copy

package com.tgb.jms.ActiveMQ;  

  

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;  

  

/** 

 * 消息消费者 

 *  

 * @author wangzhipeng 

 *  

 */  

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(false, Session.AUTO_ACKNOWLEDGE);// 第一个参数为是否开启事务  

            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 (Exception e) {  

            e.printStackTrace();  

        }  

    }  

}  

       然后我们运行JMSProducer的main方法来生产10条消息，成功执行后，我们刷新一下后台页面，可以看到我们刚才创建的消息队列：FirstQueue1，消息个数为10，没有消费者，进入队列的消息为10，出列消息个数为0：

　　


       然后我们运行JMSConsumer类的main方法，来接收FirstQueue1的消息，成功执行后，我们再刷新后台页面，发现消息已被接收：

　　

　　P2P开发模式优化

　　上面的实例中我们接收消息的方式为，写了一个死循环while (true) 来不停的接收消息，这样很浪费cpu资源，实际生产中不会这么做。这里，我们可以注册一个监听器，当监听到有消息入队列后，才去接收消息。所以，改动如下：

　　首先，新建一个监听类Listener，需要实现MessageListener接口：

[java] view
plain copy

package com.tgb.jms.ActiveMQ2;  

  

import javax.jms.JMSException;  

import javax.jms.Message;  

import javax.jms.MessageListener;  

import javax.jms.TextMessage;  

  

/** 

 * 消息监听 

 *  

 * @author zhipeng 

 *  

 */  

public class Listener implements MessageListener {  

  

    @Override  

    public void onMessage(Message message) {  

        // TODO Auto-generated method stub  

        try {  

            System.out.println("收到的消息：" + ((TextMessage) message).getText());  

        } catch (JMSException e) {  

            // TODO Auto-generated catch block  

            e.printStackTrace();  

        }  

    }  

  

}  

　　然后我们将消息接收者类JMSConsumer的while (true)死循环代码替换为messageConsumer.setMessageListener

(newListener());即可，如下图：

　　


　　这样一来我们就符合我们人为的思维操作了，会节省很多cpu资源。

五、    PubSub开发模式实战

       P2P的模式相当于一个消息生产者一个消费者，PubSub中可以有一个消息发布者多个消息订阅者，开发与P2P几乎一样，只是将createQueue（创建消息队列）改为了createTopic（创建主题）；生产消息改为了发布消息；接收消息改为了订阅消息。代码这里就不再贴出来了。有需要的同志到这里下载http://download.csdn.net/detail/wang379275614/9023177。

       这里需要注意的是，订阅者必须先订阅，然后发布者发送消息后，订阅者才能自动收到消息。

六、    总结

       ActiveMQ是JMS规范的一个实现产品，主要用在两个程序之间，或分布式系统中发送消息，进行异步通信。我们可以用它来解决高并发的问题，或者分布式事务的问题等。