P2P

P2P

关注
发信
关注(28)粉丝(399)

activeMQ "HelloWorld"实现

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

开发环境

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

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

项目建立与实现

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

activeMQ "HelloWorld"实现-LMLPHP

操作步骤

  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，终端打印出如下信息：

activeMQ &quot;HelloWorld&quot;实现-LMLPHP

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

activeMQ &quot;HelloWorld&quot;实现-LMLPHP

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

activeMQ &quot;HelloWorld&quot;实现-LMLPHP

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

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

activeMQ &quot;HelloWorld&quot;实现-LMLPHP

　　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文件，终端打印出如下信息：

activeMQ &quot;HelloWorld&quot;实现-LMLPHP

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

activeMQ &quot;HelloWorld&quot;实现-LMLPHP

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

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

activeMQ &quot;HelloWorld&quot;实现-LMLPHP

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

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

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

activeMQ &quot;HelloWorld&quot;实现-LMLPHP

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

　　源码：activeMQHelloWorld.zip

参考文献

  1. ubuntu下简单使用activemq
05-26 08:06
Powered by LMLPHP ©2025 P2P 0.048092