ActiveMQ發(fā)布與訂閱示例

發(fā)布訂閱和點(diǎn)對(duì)點(diǎn)的代碼基本相同，只是修改一下目的地，myQueue改為myTopic，以及在創(chuàng)建目的地的時(shí)候，將createQueue改為createTopic。

1、在com.bjpowernode.activemq.send包下編寫一個(gè)消息發(fā)布者TopicPublisher發(fā)送消息

package com.bjpowernode.activemq.send;
import org.apache.activemq.ActiveMQConnectionFactory;
import javax.jms.*;
public class TopicPublisher {
    public static final String BROKER_URL = "tcp://192.168.235.128:61616";
    //相當(dāng)于一個(gè)數(shù)據(jù)庫(kù)
    public static final String DESTINATION = "myTopic";
    public static void main(String[] args) {
        sendMessage();
    }
    public static void sendMessage(){
        //1 .創(chuàng)建一個(gè)連接工廠
        ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(BROKER_URL);
        Connection connection = null;
        Session session = null;
        MessageProducer messageProducer = null;
        try {
            //2. 獲取一個(gè)連接
            connection = connectionFactory.createConnection();
            //3. 創(chuàng)建一個(gè)Session 第一個(gè)參數(shù)：是否是事務(wù)消息 第二個(gè)參數(shù)：消息確認(rèn)機(jī)制（自動(dòng)確認(rèn)還是手動(dòng)確認(rèn)）
            session = connection.createSession(Boolean.FALSE, Session.AUTO_ACKNOWLEDGE);
            //4. 有了session之后，就可以創(chuàng)建消息，目的地，生產(chǎn)者和消費(fèi)者
            Message message = session.createTextMessage("Hello ActiveMQ");
            //目的地
            Destination destination = session.createTopic(DESTINATION);
            //生產(chǎn)者
            messageProducer = session.createProducer(destination);
            //發(fā)消息 沒有返回值，是非阻塞的
            messageProducer.send(message);
        } catch (JMSException e) {
            e.printStackTrace();
        }finally{
            try {
                if(messageProducer != null){
                    messageProducer.close();
                }
                if(session != null){
                    session.close();
                }
                if(connection != null){
                    connection.close();
                }
            }catch (JMSException e) {
                e.printStackTrace();
            }
        }
    }
}

2、在com.bjpowernode.activemq.receive包下編寫一個(gè)消息訂閱者TopicSubcriber接收消息

package com.bjpowernode.activemq.receive;

import org.apache.activemq.ActiveMQConnectionFactory;

import javax.jms.*;
public class TopicSubcriber {
    public static final String BROKER_URL = "tcp://192.168.235.128:61616";
    //相當(dāng)于一個(gè)數(shù)據(jù)庫(kù)（其實(shí)是一個(gè)隊(duì)列）
    public static final String DESTINATION = "myTopic";
    public static void main(String[] args) {
        receiveMessage();
    }
    public static void receiveMessage(){
        //1 .創(chuàng)建一個(gè)連接工廠
        ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(BROKER_URL);
        Connection connection = null;
        Session session = null;
        MessageConsumer messageConsumer = null;
        try {
            //2. 獲取一個(gè)連接
            connection = connectionFactory.createConnection();
            //接收消息，需要將連接啟動(dòng)一下，才可以接收到消息
            connection.start();
            //3. 創(chuàng)建一個(gè)Session 第一個(gè)參數(shù)：是否是事務(wù)消息 第二個(gè)參數(shù)：消息確認(rèn)機(jī)制（自動(dòng)確認(rèn)還是手動(dòng)確認(rèn)）
            session = connection.createSession(Boolean.FALSE, Session.AUTO_ACKNOWLEDGE);
            //4. 有了session之后，就可以創(chuàng)建消息，目的地，生產(chǎn)者和消費(fèi)者
            //目的地
            Destination destination = session.createTopic(DESTINATION);
            //消費(fèi)者
            messageConsumer = session.createConsumer(destination);
            //循環(huán)接收消息
            while (true){
                //接收消息 有返回值，是阻塞的
                Message message = messageConsumer.receive();
                //判斷消息類型
                if(message instanceof TextMessage){
                    String text = ((TextMessage) message).getText();
                    System.out.println(text);
                }
            }
        } catch (JMSException e) {
            e.printStackTrace();
        }finally{
            try {
                if(messageConsumer != null){
                    messageConsumer.close();
                }
                if(session != null){
                    session.close();
                }
                if(connection != null){
                    connection.close();
                }
            }catch (JMSException e) {
                e.printStackTrace();
            }
        }
    }
}

3、消息訂閱者先運(yùn)行，然后再運(yùn)行消息發(fā)布者