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

时间：2023-03-10 04:35:17
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

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

消息发布者的定义： 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

　　启动结果：

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

2、启动消息订阅者01

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

　　启动结果：

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

3、启动消息订阅者02

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

　　启动结果：

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

4、启动消息发布者

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

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

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

　　订阅者01读取消息后：

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

　　订阅者02读取消息后：

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

相关文章