activeMQ "HelloWorld"实现

本文主要介绍activeMQ在应用程序中是如何使用的，同个两个实例进行说明，这两个实例分别针对P2P模式和Pub/Sub模式。

开发环境

• 操作系统：Ubuntu 16.10
• 开发平台：Eclipse Neon Release (4.6.0)
• ActiveMQ版本：apache-activemq-5.14.3

　　具体的环境下载与配置这里就不在详细描述啦

项目建立与实现

　　先为大家展示以下项目最后的结构图：

操作步骤

1. 在Eclipse中新建一个最基本的Java Project，本项目命名为“activeMQHelloWorld”
2. 在项目根目录下建立文件夹libs，并将activemq-all-5.14.3.jar依赖包复制到文件夹中
3. 通过 JavaBuildPath 中的libraries将依赖包引入项目中

　　到目前位置项目框架搭建完毕（简单容易吧）

　　分别实现P2P消息模型和Pub/Sub消息模型，首先实现P2P消息模型：

　　P2P消息模型

　　　　编写生产者代码QueueProducer.java如下：

package com.unionpay.activemq;

import javax.jms.Connection;
import javax.jms.ConnectionFactory;
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 QueueProducer {
    
    //默认连接用户
    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(USERNAME, PASSWORD, BROKEURL);
        
        try{
            //通过连接工厂获取连接
            connection = connectionFactory.createConnection();
            //启动连接
            connection.start();
            //创建session,第一个参数true表示支持事物，false表示不支持事物，Session.AUTO_ACKKNOWLEDGE
            //表示自动确认，客户端发送和接受消息不需要做额外的工作
            session = connection.createSession(true, Session.AUTO_ACKNOWLEDGE);
            //创建一个名为HelloWorld的消息队列
            destination = session.createQueue("QueueTest");
            //创建消息生产者
            messageProducer = session.createProducer(destination);
            //发送消息
            sendMessage(session,messageProducer);
            //提交消息
            session.commit();
            session.close();
        }catch(Exception e){
            e.printStackTrace();
        }finally{
            if(connection != null){
                try{
                    connection.close();
                }catch(Exception e){
                    e.printStackTrace();
                }
            }
        }
    }
    
    /**
     * 发送消息
     * @param session
     * @param messageProducer
     * @throws Exception
     */
    public static void sendMessage(Session session,MessageProducer messageProducer) throws Exception{
        for(int i=0;i<SENDNUM;i++){
            //创建一条文笔消息
            TextMessage message = session.createTextMessage("ActiveMQ 发送消息"+ i);
            System.out.println("发送消息：Activemq发送消息" + i);
            
            messageProducer.send(message);
        }
    }
}

　　编写消费者代码QueueConsumer.java代码如下：

package com.unionpay.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 jxwch
 *
 */
public class QueueConsummer {

    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 = new ActiveMQConnectionFactory(USERNAME, PASSWORD, BROKEURL);
        try {
            Connection connection = connectionFactory.createConnection();

            connection.start();

            Session session = connection.createSession(true, Session.AUTO_ACKNOWLEDGE);

            Destination destination = session.createQueue("QueueTest");

            MessageConsumer messageConsumer = session.createConsumer(destination);

            while (true) {
                //100000代表100000毫秒
                TextMessage message = (TextMessage) messageConsumer.receive(100000);
                if (message != null) {
                    System.out.println("收到消息：" + message.getText());
                } else {
                    break;
                }
            }
        } catch (JMSException e) {
            e.printStackTrace();
        }
    }

}

　　到此，P2P模型代码已经全部编写完成，可以测试喽：

　　当然，我们要测试activeMQ，那么首先一定要启动服务器：

cd apache-activemq-5.14.3/bin
bash activemq start

　　通过访问自带监控应用查看服务器是否启动正常：http://127.0.0.1:8161/admin/

　　若服务器运行正常，首先在Eclipse中运行QueueProducer.java，终端打印出如下信息：

　　此时查看监控程序页面，点击“Queue”出现如下信息：

　　从截图中我们可以看到，在Queue消息中，有一个Name为QueueTest的消息队列，其中“Number Of Pending Message”表示队列中存在10条消息，“Message Enqueued” 表示有10条消息正在排队。通过点击Views中的Browser可以查看队列中的消息：

　　并且可以通过Delete对这些消息进行删除操作。

　　下面我们继续运行QueueConsumer.java，终端打印如下信息：

　　Pub/Sub 模型

　　首先编写Publisher端文件TopicProducer.java:

package com.unionpay.activemq;

import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.DeliveryMode;
import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.MapMessage;
import javax.jms.MessageProducer;
import javax.jms.Session;
import javax.jms.Topic;

import org.apache.activemq.ActiveMQConnection;
import org.apache.activemq.ActiveMQConnectionFactory;

public class TopicProducer {

    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 = new ActiveMQConnectionFactory(USERNAME, PASSWORD, BROKEURL);
        Connection connection = null;
        Session session = null;
        Topic  topic = null;
        MessageProducer messageProducer = null;
        try {
            connection = connectionFactory.createConnection();

            connection.start();

            session = connection.createSession(true, Session.AUTO_ACKNOWLEDGE);

            topic = session.createTopic("NEWS");

            messageProducer = session.createProducer(topic);
            
            messageProducer.setDeliveryMode(DeliveryMode.PERSISTENT);

            for (int i = 0; i < SENDNUM; i++) {
                MapMessage mapMessage = session.createMapMessage();

                mapMessage.setLong("count", i);

                messageProducer.send(mapMessage);

                System.out.println("发布者发布消息：" + i);

                session.commit();
            }
        } catch (JMSException e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        } finally {
            if (session != null) {
                try {
                    session.close();
                } catch (JMSException e) {
                    e.printStackTrace();
                }
            }

            if (connection != null) {
                try {
                    connection.close();
                } catch (JMSException e) {
                    e.printStackTrace();
                }
            }
        }
    }
}

　　然后编辑Subscriber端文件TopicConsumer.java：

package com.unionpay.activemq;

import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.JMSException;
import javax.jms.MapMessage;
import javax.jms.Message;
import javax.jms.MessageConsumer;
import javax.jms.MessageListener;
import javax.jms.Session;
import javax.jms.Topic;

import org.apache.activemq.ActiveMQConnection;
import org.apache.activemq.ActiveMQConnectionFactory;

public class TopicConsumer {

    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 = new ActiveMQConnectionFactory(USERNAME, PASSWORD, BROKEURL);
        
        Connection connection = null;
        try {
            connection = connectionFactory.createConnection();
            
            connection.start();
            
            Session session = connection.createSession(true, Session.AUTO_ACKNOWLEDGE);
            
            Topic topic  = session.createTopic("NEWS");
            
            MessageConsumer messageConsumer = session.createConsumer(topic);
            
            while(true){
                
                MessageListener messageListener = new MessageListener(){

                    @Override
                    public void onMessage(Message message) {
                        // TODO Auto-generated method stub
                        MapMessage mapMessage = null;
                        try{
                            mapMessage = (MapMessage)message;
                            
                            System.out.println("Receiver Message:" + mapMessage.getLong("count"));
                        }catch(JMSException e){
                            e.printStackTrace();
                        }                    
                    }    
                };
                messageConsumer.setMessageListener(messageListener);
            }
            
            
        } catch (JMSException e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        }finally{
            if(connection != null){
                try{
                    connection.close();
                }catch(JMSException e){
                    e.printStackTrace();
                }
            }
        }
    }

}

　　到此，Pub/Sub模型代码编写完毕，下面运行TopicProducer.java文件，终端打印出如下信息：

　　然后查看监控程序，点击“Topics”：

　　从截图中我们可以看出Name中多了一个NEWS主题，并且Messages Enqueued为10。

　　然后运行客户端TopicConsumer.java文件，终端显示如下：

　　从截图中我们并没有发现客户端消费了消息，这是为啥呢？

　　因为在Pub/Sub模型中，发布者和订阅者有时间上的依赖性，针对某个主题，必须先创建订阅者，然后才能发布消息，这样才能保证订阅者可以收到消息。

　　重新运行TopicConsumer.java文件，就可以看见消费信息了：

　　至此，两种模式已经全部介绍完毕。

　　源码：activeMQHelloWorld.zip

参考文献

1. ubuntu下简单使用activemq