ActiveMQ入門系列三：發佈/訂閱模式

時間 2019-11-08

原文原文鏈接

在上一篇《ActiveMQ入門系列二：入門代碼實例（點對點模式）》中提到了ActiveMQ中的兩種模式：點對點模式（PTP）和發佈/訂閱模式（Pub & Sub），詳細介紹了點對點模式並用代碼實例進行說明，今天就介紹下發布/訂閱模式。html

1、理論基礎java

發佈/訂閱模式的工做示意圖：apache

消息生產者將消息（發佈）到topic中，能夠同時有多個消息消費者（訂閱）消費該消息。
和點對點方式不一樣，發佈到topic的消息會被全部訂閱者消費。
當生產者發佈消息，無論是否有消費者，都不會保存消息。
必定要先有消息的消費者，後有消息的生產者。

2、代碼實現session

生產者

package com.sam.topic;

import org.apache.activemq.ActiveMQConnectionFactory;

import javax.jms.*;

/**
 * @author JAVA開發老菜鳥
 *
 */
public class TopicProducer {

    public static final  String QUEUE_NAME = "topic-demo";//隊列名

    public void producer(String message) throws JMSException {
        ConnectionFactory factory = null;
        Connection connection = null;
        Session session = null;
        MessageProducer producer = null;
        try {
            /**
             * 1.建立鏈接工廠
             * 建立工廠，構造方法有三個參數：分別是用戶名、密碼、鏈接地址
             * 無參構造：有默認的鏈接地址，localhost
             * 一個參數：無驗證模式，無用戶的認證
             * 三個參數：有認證和鏈接地址
             */
            factory = new ActiveMQConnectionFactory("admin","admin","tcp://localhost:61616");
            /**
             * 2.建立鏈接
             * 無參數
             * 有參數：用戶名、密碼
             */
            connection = factory.createConnection();
            /**
             * 3.啓動鏈接
             * 生產者能夠不啓動，由於在發送消息的時候回進行檢查
             * 若是未啓動鏈接，會自動啓動
             * 若是有特殊配置，須要配置完成後再啓動鏈接
             */
            connection.start();
            /**
             * 4.用鏈接建立會話
             * 有兩個參數：是否須要事務、消息確認機制
             * 若是支持事務，對於生產者來講第二個參數就無效了，建議傳入Session.SESSION_TRANSACTED
             * 若是不支持事務，第二個參數必須傳遞且有效
             *
             * AUTO_ACKNOWLEDGE：自動確認，消息處理後自動確認（商業開發不推薦）
             * CLIENT_ACKNOWLEDGE：客戶端手動確認，消費者處理後必須手動確認
             * DUPS_OK_ACKNOWLEDGE：有副本的客戶端手動確認，消息能夠屢次處理（不建議）
             */
            session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
            /**
             * 5.用會話建立目的地（主題）、生產者、消息
             * 隊列名是隊列的惟一標記
             * 建立生產者的時候能夠不指定目的地，能夠在發送的時候指定
             */
            Destination destination = session.createTopic(QUEUE_NAME);
            producer = session.createProducer(destination);
            TextMessage textMessage = session.createTextMessage(message);
            /**
             * 6.生產者發送消息到目的地
             */
            producer.send(textMessage);
            System.out.println("消息發送成功");
        } catch(Exception ex){
            throw ex;
        } finally {
            /**
             * 7.釋放資源
             */
            if(producer != null){
                producer.close();
            }

            if(session != null){
                session.close();
            }

            if(connection != null){
                connection.close();
            }
        }
    }

    public static void main(String[] args){
        TopicProducer producer = new TopicProducer();
        try{
            producer.producer("hello, activemq");
        } catch (Exception ex){
            ex.printStackTrace();
        }
    }
}

發佈/訂閱模式的生產者和點對點模式的代碼主要區別就是Destination的建立方式，點對點模式是調用session.createQueue(QUEUE_NAME)，而發佈/訂閱模式是調用session.createTopic(QUEUE_NAME)。tcp

消費者

package com.sam.topic;

import org.apache.activemq.ActiveMQConnectionFactory;

import javax.jms.*;
import java.io.IOException;

/**
 * @author JAVA開發老菜鳥
 *
 * 觀察者消費--監聽消費
 */
public class TopicConsumer {

    public void consumer() throws JMSException, IOException {
        ConnectionFactory factory = null;
        Connection connection = null;
        Session session = null;
        MessageConsumer consumer = null;
        try {
            factory = new ActiveMQConnectionFactory("admin","admin","tcp://localhost:61616");
            connection = factory.createConnection();
            /**
             * 消費者必須啓動鏈接，不然沒法消費
             */
            connection.start();
            session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
            Destination destination = session.createTopic(TopicProducer.QUEUE_NAME);
            consumer = session.createConsumer(destination);
            /**
             * 註冊監聽器，隊列中的消息變化會自動觸發監聽器，接收並自動處理消息
             *
             * 監聽器一旦註冊，永久有效，一直到程序關閉
             * 監聽器能夠註冊多個，至關於集羣
             * activemq自動輪詢多個監聽器，實現並行處理
             */
            consumer.setMessageListener(new MessageListener() {
                @Override
                public void onMessage(Message message) {

                    try {
                        TextMessage om = (TextMessage) message;
                        String data = om.getText();
                        System.out.println(data);
                    } catch (JMSException e) {
                        e.printStackTrace();
                    }
                }
            });
        } catch(Exception ex){
            throw ex;
        }
    }

    public static void main(String[] args){
        TopicConsumer consumer = new TopicConsumer();
        try{
            consumer.consumer();
        } catch (Exception ex){
            ex.printStackTrace();
        }
    }
}