您的位置:首页 > 其它

消息队列-ActiveMQ学习笔记(三)-发布-订阅消息模式实现

2017-11-03 11:09 861 查看
发布-订阅消息模式与点对点模式类似,只不过在session创建消息队列时,由session.createQuene()变为session.createTopic()。



消息发布者代码:

[java] view
plain copy

package com.feiyang.activemq2;  

  

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 MCL 

 * 

 */  

public class JMSProducer {  

    private static final String USERNAME = ActiveMQConnectionFactory.DEFAULT_USER;// 默认的连接的用户名  

    private static final String PASSWORD = ActiveMQConnectionFactory.DEFAULT_PASSWORD;// 默认的连接密码  

    private static final String BROKEURL = ActiveMQConnectionFactory.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(JMSProducer.USERNAME, JMSProducer.PASSWORD,  

                JMSProducer.BROKEURL);  

  

        try {  

            // 通过连接工厂获取连接  

            connection = connectionFactory.createConnection();  

            // 启动连接  

            connection.start();  

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

            destination = session.createTopic("FirstTopic");// 创建消息队列  

            messageProducer = session.createProducer(destination);// 创建消息生产者  

            sendMessage(session, messageProducer);  

            // 由于设置添加事务,这里需要使用提交才能将数据发送出去  

            session.commit();  

  

        } catch (JMSException e) {  

            // TODO Auto-generated catch block  

            e.printStackTrace();  

        } finally {  

            if (connection != null) {  

                try {  

                    connection.close();  

                } catch (JMSException e) {  

                    // TODO Auto-generated catch block  

                    e.printStackTrace();  

                }  

            }  

        }  

    }  

  

    // 发送消息  

    public static void sendMessage(Session session, MessageProducer messageProducer) {  

        for (int i = 0; i < JMSProducer.SENDNUM; i++) {  

            try {  

                TextMessage message = session.createTextMessage("Active MQ发送消息" + i);  

                System.out.println("发布消息:Active MQ发送消息");  

                messageProducer.send(message);  

            } catch (JMSException e) {  

                // TODO Auto-generated catc h block  

                e.printStackTrace();  

            }  

        }  

    }  

}  

消息订阅者代码:

[java] view
plain copy

package com.feiyang.activemq2;  

  

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 org.apache.activemq.ActiveMQConnection;  

import org.apache.activemq.ActiveMQConnectionFactory;  

  

/** 

 * 消息订阅者一 

 * @author MCL 

 * 

 */  

public class JMSConsumer {  

    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; // 连接工厂  

        Connection connection = null; // 连接  

        Session session; // 会话 接受或者发送消息的线程  

        Destination destination; // 消息的目的地  

        MessageConsumer messageConsumer; // 消息的消费者  

          

  

        connectionFactory = new ActiveMQConnectionFactory(JMSConsumer.USERNAME, JMSConsumer.PASSWORD,  

                JMSConsumer.BROKEURL);  

        try {  

            // 通过连接工厂获取连接  

            connection=connectionFactory.createConnection();  // 通过连接工厂获取连接  

            connection.start(); // 启动连接  

            session=connection.createSession(Boolean.FALSE, Session.AUTO_ACKNOWLEDGE); // 创建Session  

            destination=session.createTopic("FirstTopic");   

            messageConsumer=session.createConsumer(destination); // 创建消息消费者  

            messageConsumer.setMessageListener(new Listener()); // 注册消息监听  

              

        } catch (JMSException e) {  

            // TODO Auto-generated catch block  

            e.printStackTrace();  

        }   

    }  

}  

监听器代码:

[java] view
plain copy

package com.feiyang.activemq2;  

  

import javax.jms.JMSException;  

import javax.jms.Message;  

import javax.jms.MessageListener;  

import javax.jms.TextMessage;  

/** 

 * 消息监听器 

 * @author MCL 

 * 

 */  

public class Listener implements MessageListener{  

  

    @Override  

    public void onMessage(Message message) {  

        // TODO Auto-generated method stub  

        try {  

            System.out.println("消息订阅者一收到的消息:"+((TextMessage)message).getText());  

        } catch (JMSException e) {  

            // TODO Auto-generated catch block  

            e.printStackTrace();  

        }  

    }  

  

}  

我们可以定义多个消息订阅者及其监听器。这里定义了两个订阅者,由于代码相似所以只粘贴一份。

由于发布-订阅模型的关系,需要先进行订阅后,才能接收发布者的消息。

先启动 订阅者一和订阅者二的线程,然后用发布者 发布消息。打开后台管理界面,点击Topics



有上述图片可以看出,FirstTopic中,消息发布者发布了10条信息,并由两个订阅者进行消费,每人消费10条信息。
内容来自用户分享和网络整理,不保证内容的准确性,如有侵权内容,可联系管理员处理 点击这里给我发消息
标签:  jms activemq