說明

基于ActiveMQ學(xué)習(xí)（二），學(xué)習(xí)ActiveMQ的另一種消息模式Topic訂閱模式，同時(shí)將普通的文本消息改為對(duì)象消息。

代碼

demo代碼

訂閱模式

文件目錄

• 編寫被收發(fā)的對(duì)象 UserModel

package com.sima.topics;
import java.io.Serializable;
/**
 * Created by Maple on 2017-05-28.
 */
public class UserModel implements Serializable {

    private String name;
    private int age;

    public UserModel() {
    }
    public UserModel(String name, int age) {
        this.name = name;
        this.age = age;
    }
    public String getName() {
        return name;
    }
    public void setName(String name) {
        this.name = name;
    }

    public int getAge() {
        return age;
    }

    public void setAge(int age) {
        this.age = age;
    }

    @Override
    public String toString() {
        return "UserModel{" +
                "name='" + name + '\'' +
                ", age=" + age +
                '}';
    }
}

注：被收發(fā)的對(duì)象類需要實(shí)現(xiàn)Serializable

• 編寫消息生產(chǎn)者

package com.sima.topics;
import org.apache.activemq.ActiveMQConnection;
import org.apache.activemq.ActiveMQConnectionFactory;
import org.apache.activemq.command.ActiveMQObjectMessage;
import javax.jms.*;
/**
 * Created by Maple on 2017-05-28.
 */
public class JMSProducer {
    private static final String USERNAME = ActiveMQConnection.DEFAULT_USER; // 默認(rèn)的連接用戶名
    private static final String PASSWORD = ActiveMQConnection.DEFAULT_PASSWORD; // 默認(rèn)的連接密碼
    private static final String BROKEURL = ActiveMQConnection.DEFAULT_BROKER_URL; // 默認(rèn)的連接地址
//  private static final String BROKEURL= "tcp://localhost:8161"; // 默認(rèn)的連接地址
    private static final int SENDNUM = 3; // 發(fā)送的消息數(shù)量
    public static void  main(String[] args){
//        ConnectionFactory connectionFactory; // 連接工廠
        ActiveMQConnectionFactory connectionFactory; // 連接工廠
        Connection connection = null; // 連接
        Session session; // 會(huì)話 接受或者發(fā)送消息的線程
        Destination destination; // 消息的目的地
        MessageProducer messageProducer; // 消息生產(chǎn)者
        connectionFactory = new ActiveMQConnectionFactory(JMSProducer.USERNAME,
                JMSProducer.PASSWORD, JMSProducer.BROKEURL);
        connectionFactory.setTrustAllPackages(true);
        try {
            connection = connectionFactory.createConnection(); // 通過連接工廠獲取連接
            connection.start(); // 啟動(dòng)連接
            session = connection.createSession(Boolean.TRUE, Session.AUTO_ACKNOWLEDGE); // 創(chuàng)建Session
//            destination = session.createQueue("TestQueueFirst"); // 創(chuàng)建消息隊(duì)列
            destination = session.createTopic("TestTopicFirst"); // 創(chuàng)建消息隊(duì)列
            messageProducer = session.createProducer(destination); // 創(chuàng)建消息生產(chǎn)者
            //設(shè)置不持久化
//            messageProducer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);
            sendMessage(session, messageProducer); // 發(fā)送消息
            session.commit();
        } catch (JMSException e) {
            e.printStackTrace();
        }finally {
            if(connection!=null){
                try {
                    connection.close();
                } catch (JMSException e) {
                    e.printStackTrace();
                }
            }
        }
    }

    private static void sendMessage(Session session, MessageProducer messageProducer) {
        for(int i=0;i< JMSProducer.SENDNUM;i++){
            TextMessage message= null;
            try {
//                message = session.createTextMessage("ActiveMQ 發(fā)送的消息-"+i);
//                System.out.println("發(fā)送消息：" + "ActiveMQ 發(fā)送的消息-" + i);
//                messageProducer.send(message);

                UserModel userModel = new UserModel("maple", 28);
                ObjectMessage objectMessage = session.createObjectMessage(userModel);
                objectMessage.setObject(userModel);
//                ActiveMQObjectMessage
                System.out.println("發(fā)送消息：" + "ActiveMQ 發(fā)送的消息-" + userModel.toString());
                messageProducer.send(objectMessage);
            } catch (JMSException e) {
                e.printStackTrace();
            }
        }
    }
}

• 消息消費(fèi)者

package com.sima.topics;
import com.sima.queues.MyListener;
import org.apache.activemq.ActiveMQConnection;
import org.apache.activemq.ActiveMQConnectionFactory;
import javax.jms.*;
/**
 * Created by Maple on 2017-05-28.
 */
public class JMSConsumerByListener1 {
    private static final String USERNAME = ActiveMQConnection.DEFAULT_USER; // 默認(rèn)的連接用戶名
    private static final String PASSWORD = ActiveMQConnection.DEFAULT_PASSWORD; // 默認(rèn)的連接密碼
    private static final String BROKEURL = ActiveMQConnection.DEFAULT_BROKER_URL; // 默認(rèn)的連接地址
//    private static final String BROKEURL= "tcp://localhost:8161"; // 默認(rèn)的連接地址
    public static void main(String[] args){
//        ConnectionFactory connectionFactory; // 連接工廠
        ActiveMQConnectionFactory connectionFactory; // 連接工廠
        Connection connection = null; // 連接
        Session session; // 會(huì)話 接受或者發(fā)送消息的線程
        Destination destination; // 消息的目的地
        MessageConsumer messageConsumer; // 消息的消費(fèi)者
        // 實(shí)例化連接工廠
        connectionFactory=new ActiveMQConnectionFactory(JMSConsumerByListener1.USERNAME, JMSConsumerByListener1.PASSWORD, JMSConsumerByListener1.BROKEURL);
        connectionFactory.setTrustAllPackages(true);
        try {
            connection=connectionFactory.createConnection();  // 通過連接工廠獲取連接
            connection.start(); // 啟動(dòng)連接
            session=connection.createSession(Boolean.FALSE, Session.AUTO_ACKNOWLEDGE); // 創(chuàng)建Session
//            destination=session.createQueue("TestQueueFirst");  // 創(chuàng)建連接的消息隊(duì)列
            destination=session.createTopic("TestTopicFirst");  // 創(chuàng)建連接的消息隊(duì)列
            messageConsumer=session.createConsumer(destination); // 創(chuàng)建消息消費(fèi)者
//            messageConsumer.setMessageListener(new MyListener());// 注冊(cè)消息監(jiān)聽
            messageConsumer.setMessageListener(new MyObjectListener());// 注冊(cè)消息監(jiān)聽
        } catch (JMSException e) {
            e.printStackTrace();
        }
    }
}

• 監(jiān)聽類

package com.sima.topics;
import javax.jms.*;
/**
 * Created by Maple on 2017-05-28.
 */
public class MyObjectListener implements MessageListener {
    public void onMessage(Message message) {
        try {
            UserModel u = (UserModel) ((ObjectMessage) message).getObject();
            System.out.println("通過MyListener收到的ObjectMessage消息："+ u.toString());
        } catch (JMSException e) {
            e.printStackTrace();
        }
    }
}

• 先運(yùn)行消費(fèi)者類進(jìn)行消息訂閱，查看 http://localhost:8161/admin/subscribers.jsp

訂閱者

注冊(cè)成消費(fèi)者

• 運(yùn)行消息生產(chǎn)者，此時(shí)消費(fèi)者會(huì)收到通知并進(jìn)行消費(fèi)

消息訂閱消費(fèi)

• 當(dāng)有多個(gè)訂閱者的情況下，每個(gè)訂閱者都將收到相同數(shù)量的消息

白名單

ActiveMQ自5.12.2版本之后，強(qiáng)制用戶指定一份可信任的packages白名單，以對(duì)付ObjectMessage存在的安全漏洞。具體內(nèi)容可參考：http://activemq.apache.org/objectmessage.html。
會(huì)遇到如下問題：

javax.jms.JMSException: Failed to build body from content. Serializable class not available to broker. Reason: java.lang.ClassNotFoundException: Forbidden class com.sima.topics.UserModel! 
This class is not trusted to be serialized as ObjectMessage payload. 
Please take a look at http://activemq.apache.org/objectmessage.html for more information on how to configure trusted classes.

官方提供的白名單配置方法如下：

白名單配置