ActiveMQ的发布者/订阅者模型示例

Posted 阿叮339

tags:

篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了ActiveMQ的发布者/订阅者模型示例相关的知识，希望对你有一定的参考价值。

ActiveMQ的发布者/订阅者模型入门示例

（1）下载安装activemq，启动activeMQ。

　　详细步骤参考博客：http://www.cnblogs.com/DFX339/p/9050878.html

 　　

（2）创建maven项目，java项目或者web项目都可以。

 项目源码下载地址：https://github.com/DFX339/activeMQ_demo.git

 目录结构如下：（queue包下的是activemq队列模型的入门示例，需要的可以参考 http://www.cnblogs.com/DFX339/p/9050950.html）

 需要编写的文件：MQProducer.java    Listener01.java     MQConsumer01.java     Listener02.java     MQConsumer02.java    pom.xml

 

消息发布者的定义： MQProducer.java

主要步骤：

　　/**
         * 1.创建连接工厂
         * 2.创建连接实例
         * 3、启动连接
         * 4、创建session创建接收或发送的线程实例（创建session的时候定义是否要启用事务，
         * 且事务类型是Auto_ACKNOWLEDGE也就是消费者成功在Listern中获得消息返回时，会话自动确定用户收到消息）
         * 5、创建队列（消息发送的目的地）
         * 6、创建消息发送者
         * 7、创建消息
         * 8、发送消息
         * 9、session.commit();提交千万不要忘记了
         */
 代码如下：

package cn.dfx.activeMQ_demo.topic;

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.ActiveMQConnectionFactory;

/**
 * 订阅消息的发送消息方
 * @author Administrator
 *
 */
public class MQProducer {
    
    public static void main(String[] args){
        
        /**
         * 1.创建连接工厂
         * 2.创建连接实例
         * 3、启动连接
         * 4、创建session创建接收或发送的线程实例（创建session的时候定义是否要启用事务，
         * 且事务类型是Auto_ACKNOWLEDGE也就是消费者成功在Listern中获得消息返回时，会话自动确定用户收到消息）
         * 5、创建队列（消息发送的目的地）
         * 6、创建消息发送者
         * 7、创建消息
         * 8、发送消息
         * 9、session.commit();提交千万不要忘记了 
         */
        
        ConnectionFactory connFactory = null;
        Connection conn = null;
        Session session = null;
        Destination destination = null;
        
        //连接参数定义： 用户名 密码 url 
        String name = "system";
        String password = "manager";
        String url = "failover://tcp://localhost:61616";
        
        System.out.println("消息发布者开始发布消息了……");
        try{
            
            //创建连接工厂
            //这里的连接参数可以使用常量：Constants.MQ_NAME, Constants.MQ_PASSWORD, Constants.MQ_BROKETURL
            connFactory = new ActiveMQConnectionFactory(name,password,url);
            
            //通过连接工厂创建连接实例
            conn = connFactory.createConnection();
            
            //启动连接
            conn.start();
            
            //4、创建session创建接收或发送的线程实例（创建session的时候定义是否要启用事务，且事务类型是Auto_ACKNOWLEDGE也就是消费者成功在Listern中获得消息返回时，会话自动确定用户收到消息）
            //也可以使用：session = connection.createSession(Boolean.TRUE, Session.AUTO_ACKNOWLEDGE);
            session = conn.createSession(true, Session.AUTO_ACKNOWLEDGE);
            
            //创建队列，也就是消息发送的目的地
            destination = session.createTopic("FirstTopic");
            
            //创建消息发布者
            MessageProducer  messageProducer = session.createProducer(destination);
            
            //创建需要发送的消息
            TextMessage textMessage = session.createTextMessage();
            textMessage.setText("Hello，broadcast message NO.1!");
            
            //发送消息
            messageProducer.send(textMessage);
            
            //一定要记得这个，提交呀，
            session.commit();
            
            System.out.println("消息发布者："+textMessage.getText());
            
        }catch(JMSException e){
             e.printStackTrace();
        } finally {
            
            //关闭连接
            if (conn != null) {
                try {
                    conn.close();
                } catch (JMSException e) {
                    e.printStackTrace();
                }    
            }
        
        }
    
    }
}

 

消息订阅者01的定义： MQConsumer01.java

主要步骤：

　　/**
         * 1.创建连接工厂
         * 2.创建连接实例
         * 3、启动连接
         * 4、  创建接收或发送的线程实例（消费者就不需要开启事务了）
         * 5、创建队列（消息发送的目的地）
         * 6、创建消息接收者
         * 7、注册消息监听
         */
  代码示例：

package cn.dfx.activeMQ_demo.topic;

import java.io.IOException;

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.Session;
import javax.jms.TextMessage;

import org.apache.activemq.ActiveMQConnectionFactory;

/**
 * 订阅消息的消息接收方01
 * @author Administrator
 *
 */
public class MQConsumer01 {
    
    public static void main(String[] args) throws IOException{
        
        /**
         * 1.创建连接工厂
         * 2.创建连接实例
         * 3、启动连接
         * 4、  创建接收或发送的线程实例（消费者就不需要开启事务了）
         * 5、创建队列（消息发送的目的地）
         * 6、创建消息接收者
         * 7、注册消息监听
         */
        
        ConnectionFactory connFactory = null;
        Connection conn = null;
        Session session = null;
        Destination destination = null;
        
        //创建连接工厂需要的参数
        String name  = "system";
        String password = "manager";
        String url = "failover://tcp://localhost:61616";
        
        try{
            
            //创建连接工厂
            connFactory = new ActiveMQConnectionFactory(name,password,url);
            
            //创建连接实例
            conn = connFactory.createConnection();
            
            //启动连接
            conn.start();
            
            //创建session（创建接收或发送的线程实例（消费者就不需要开启事务了））
            session = conn.createSession(true, Session.AUTO_ACKNOWLEDGE);
            
            //创建消息目的地（消费者从这里读取消息）
            destination = session.createTopic("FirstTopic");
            
            //创建消费者
            MessageConsumer messageConsumer = session.createConsumer(destination);
            
            //消费者读取消息，监听消息
            messageConsumer.setMessageListener(new Listener01());
            
            System.out.println("订阅者01已经准备好接收消息！");
            
//            //8、程序等待接收用户消息
//            System.in.read();
//            //9、关闭资源
//            messageConsumer.close();
//            session.close();
//            conn.close();
            
        }catch(JMSException e){
            e.printStackTrace();
        }
    }
}

 

消息订阅者01的监听器的编写： Listener01.java

package cn.dfx.activeMQ_demo.topic;

import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageListener;
import javax.jms.TextMessage;

/**
 * 消息订阅 消息接收方的订阅者01的监听器
 * @author Administrator
 *
 */
public class Listener01 implements MessageListener {

    @Override
    public void onMessage(Message message) {
        
        try{
            System.out.println("订阅者01接收到的消息为："+((TextMessage) message).getText());
        }catch(JMSException e){
            e.printStackTrace();
        }
        
    }

}

 

 

消息订阅者02的定义： MQConsumer02.java

主要步骤：

　　/**
         * 1.创建连接工厂
         * 2.创建连接实例
         * 3、启动连接
         * 4、创建接收或发送的线程实例（消费者就不需要开启事务了）
         * 5、创建队列（消息发送的目的地）
         * 6、创建消息接收者
         * 7、注册消息监听
         */
  代码示例：

package cn.dfx.activeMQ_demo.topic;

import java.io.IOException;

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.Session;
import javax.jms.TextMessage;

import org.apache.activemq.ActiveMQConnectionFactory;

/**
 * 订阅消息的消息接收方02
 * @author Administrator
 *
 */
public class MQConsumer02 {
    
    public static void main(String[] args) throws IOException{
        
        /**
         * 1.创建连接工厂
         * 2.创建连接实例
         * 3、启动连接
         * 4、  创建接收或发送的线程实例（消费者就不需要开启事务了）
         * 5、创建队列（消息发送的目的地）
         * 6、创建消息接收者
         * 7、注册消息监听
         */
        
        ConnectionFactory connFactory = null;
        Connection conn = null;
        Session session = null;
        Destination destination = null;
        
        //创建连接工厂需要的参数
        String name  = "system";
        String password = "manager";
        String url = "failover://tcp://localhost:61616";
        
        try{
            
            //创建连接工厂
            connFactory = new ActiveMQConnectionFactory(name,password,url);
            
            //创建连接实例
            conn = connFactory.createConnection();
            
            //启动连接
            conn.start();
            
            //创建session（创建接收或发送的线程实例（消费者就不需要开启事务了））
            session = conn.createSession(true, Session.AUTO_ACKNOWLEDGE);
            
            //创建消息目的地（消费者从这里读取消息）
            destination = session.createTopic("FirstTopic");
            
            //创建消费者
            MessageConsumer messageConsumer = session.createConsumer(destination);
            
            //消费者读取消息，监听消息
            messageConsumer.setMessageListener(new Listener01());
            
            System.out.println("订阅者02已经准备好接收消息！");
           
        }catch(JMSException e){
            e.printStackTrace();
        }
    }
}

 

消息订阅者02的监听器的编写： Listener02.java

package cn.dfx.activeMQ_demo.topic;

import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageListener;
import javax.jms.TextMessage;

/**
 * 消息订阅 消息接收方的订阅者02的监听器
 * @author Administrator
 *
 */
public class Listener01 implements MessageListener {

    @Override
    public void onMessage(Message message) {
        
        try{
            System.out.println("订阅者02接收到的消息为："+((TextMessage) message).getText());
        }catch(JMSException e){
            e.printStackTrace();
        }
        
    }

}

 

（3）测试步骤

1、先启动activemq

　　启动方法：找到activemq安装目录，进入bin目录，双击activemq.bat

　　启动结果：

　　

 

2、启动消息订阅者01

　　启动方法： 打开MQConsumer01.java --> 右击 --> Run as  --> java Application

　　启动结果：

　　

 

3、启动消息订阅者02

　　启动方法： 打开MQConsumer02.java --> 右击 --> Run as  --> java Application

　　启动结果：

　　

 

4、启动消息发布者

　　启动方法： 打开MQProducer.java --> 右击 --> Run as  --> java Application

　　启动结果：(订阅者会自动读取消息，可以切换控制台查询读取情况)

　　

　　订阅者01读取消息后：

　　

 

　　订阅者02读取消息后：

　　

 

以上是关于ActiveMQ的发布者/订阅者模型示例的主要内容，如果未能解决你的问题，请参考以下文章

ActiveMQ之队列和主题发布订阅实例

3，ActiveMQ-入门（基于JMS发布订阅模型）

ActiveMQ第一个示例

activeMq初步学习

WSO2 ESB 5.0.0 配置 JMS 传输（ActiveMQ）- 主题消息发布与订阅

WSO2 ESB 5.0.0 配置 JMS 传输（ActiveMQ）- 主题消息发布与订阅