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

时间：2018-05-18 15:43:26 阅读：187 评论：0 收藏：0 [点我收藏+]

标签：queue lis password font 情况 led app run 示例

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

主要步骤：

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

}