# activeMQ 對于消息的傳遞有兩種類型： **一種是點對點的**，即一個生產者和一個消費者一一對應； **另一種是發布/** **訂閱模式**，即一個生產者產生消息并進行發送后，可以由多個消費者進行接收。 JMS定義了五種不同的消息正文格式，以及調用的消息類型，允許你發送并接收以一些不同形式的數據，提供現有消息格式的一些級別的兼容性。 - StreamMessage -- Java原始值的數據流 - MapMessage--一套名稱\-值對 - TextMessage--一個字符串對象 - ObjectMessage--一個序列化的 Java對象 - BytesMessage--一個字節的數據流 其中在開發中常用到的是TextMessage，即消息通過json數據格式傳輸，而接收數據時通過字符串對象(TextMessage)獲取。 > 使用 ``` /* * 生產者(MessageProducer) */ @Test public void queueProducerTest() throws Exception { // 創建一個ActiveMQConnectionFactory對象并傳入與activeMq通信協議地址與端口 // 用ConnectionFactory接受創建ActiveMQConnectionFactory的對象 ConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://192.168.0.151:61616"); // 通過connectionFactory創建connection Connection connection = connectionFactory.createConnection(); // 開啟connection,調用start()方法 connection.start(); // connection創建一個session對象 // param1 是否開啟分布式事務 一般不開啟（false）如果開啟則第二個參數無效 // param2 設置應答類型 手動 自動 Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); // session創建destination對象 (queue,topic) Queue queue = session.createQueue("igeek_test"); // 隊列名稱 // session創建MessageProvider對象 MessageProducer producer = session.createProducer(queue); // session創建TextMessage對象 TextMessage textMessage = session.createTextMessage("信息：我是測試發送信息！"); // 使用MessageProvider發送消息 producer.send(textMessage); // 關閉資源 producer.close(); session.close(); connection.close(); } /* * 消費者（consumer） */ @Test public void queueConsumerTest() throws Exception { // 創建一個ActiveMQConnectionFactory對象并傳入與activeMq通信協議地址與端口 // 用ConnectionFactory接受創建ActiveMQConnectionFactory的對象 ConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://192.168.0.151:61616"); // 通過connectionFactory創建connection Connection connection = connectionFactory.createConnection(); // 開啟connection,調用start()方法 connection.start(); // connection創建一個session對象 // param1 是否開啟分布式事務 一般不開啟（false）如果開啟則第二個參數無效 // param2 設置應答類型 手動 自動 Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); // session創建destination對象 (queue,topic) Queue queue = session.createQueue("igeek_test"); // 隊列名稱 // session創建一個消費者 MessageConsumer consumer = session.createConsumer(queue); // consumer設置MessageListener consumer.setMessageListener(new MessageListener() { // params message 監聽到的消息 @Override public void onMessage(Message message) { try { // 強轉為TextMessage類型 TextMessage textMessage = (TextMessage) message; //獲得消息內容 System.out.println(textMessage.getText()); } catch (JMSException e) { e.printStackTrace(); } } }); System.in.read(); consumer.close(); session.close(); connection.close(); } ``` > 與Spring框架結合使用 - 生產者(producer) ``` <!-- 真正可以產生Connection的ConnectionFactory，由對應的 JMS服務廠商提供 --> <bean id="targetConnectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory"> <property name="brokerURL" value="tcp://192.168.0.151:61616" /> </bean> <!-- Spring用于管理真正的ConnectionFactory的ConnectionFactory --> <bean id="connectionFactory" class="org.springframework.jms.connection.SingleConnectionFactory"> <!-- 目標ConnectionFactory對應真實的可以產生JMS Connection的ConnectionFactory --> <property name="targetConnectionFactory" ref="targetConnectionFactory" /> </bean> <!-- 配置生產者 --> <!-- Spring提供的JMS工具類，它可以進行消息發送、接收等 --> <bean id="jmsTemplate" class="org.springframework.jms.core.JmsTemplate"> <!-- 這個connectionFactory對應的是我們定義的Spring提供的那個ConnectionFactory對象 --> <property name="connectionFactory" ref="connectionFactory" /> </bean> <!--這個是隊列目的地，點對點的 --> <bean id="queueDestination" class="org.apache.activemq.command.ActiveMQQueue"> <constructor-arg> <value>spring-queue</value> </constructor-arg> </bean> <!--這個是主題目的地，一對多的 --> <bean id="topicDestination" class="org.apache.activemq.command.ActiveMQTopic"> <constructor-arg value="topic" /> </bean> ``` - 消費者(consumer) ``` <!-- 真正可以產生Connection的ConnectionFactory，由對應的 JMS服務廠商提供 --> <bean id="targetConnectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory"> <property name="brokerURL" value="tcp://192.168.0.151:61616" /> </bean> <!-- Spring用于管理真正的ConnectionFactory的ConnectionFactory --> <bean id="connectionFactory" class="org.springframework.jms.connection.SingleConnectionFactory"> <!-- 目標ConnectionFactory對應真實的可以產生JMS Connection的ConnectionFactory --> <property name="targetConnectionFactory" ref="targetConnectionFactory" /> </bean> <!--這個是隊列目的地，點對點的 --> <bean id="queueDestination" class="org.apache.activemq.command.ActiveMQQueue"> <constructor-arg> <value>spring-queue</value> </constructor-arg> </bean> <!--這個是主題目的地，一對多的 --> <bean id="topicDestination" class="org.apache.activemq.command.ActiveMQTopic"> <constructor-arg value="spring-topic" /> </bean> <!-- 消費者實現類 測試 --> <bean id="myMessageListener" class="com.igeek.esgobuy.search.message.MyMessageListener"></bean> <!-- 消息監聽容器 --> <bean class="org.springframework.jms.listener.DefaultMessageListenerContainer"> <property name="connectionFactory" ref="connectionFactory" /> <property name="destination" ref="queueDestination" /> <property name="messageListener" ref="myMessageListener" /> </bean> ``` 消費者監聽實現類 ``` public class MyMessageListener implements MessageListener { /* * 處理接受到的消息 */ @Override public void onMessage(Message message) { try { // 將Message強轉為TextMessage 只有當Message對象接受的是TextMessage才能成功 TextMessage textMessage = (TextMessage) message; System.out.println(textMessage.getText()); } catch (JMSException e) { e.printStackTrace(); } } } ``` * [ ] 創建消費者和生產者的消息命名應該一致 * [ ] 一對一方式(queue)，生產者的消息會等待消費者接收，再出棧。廣播方式(topic)，當消費者沒有及時監聽，會錯過生產者的消息。