1個生產者，1個消費者，使用Queue：
方式1：
生產者將消息發送到Queue中，退出；
然后運行消費者：
、
可以看到，可以接收到消息。
方式2：
先運行消費者程序：
然后運行生產者：
消費者見下圖：
1個生產者，2個消費者，使用Queue
先運行消費者1：
在運行消費者2：
接下來運行生產者：
下面是消費者消費情況：
：
總結：
1. 使用Queue時，生產者只要將Message發送到MQ服務器端，消費者就可以進行消費，而無需生產者程序一直運行；
2. 消息是按照先入先出的順序，一旦有消費者將Message消費，該Message就會從MQ服務器隊列中刪去；
3. 有文章說，“生產者”<-->"消費者"是一對一的關系，其實並不准確，從上面可以看出，一個生產者產生的消息，可以被
多個消費者進行消費，只不過多個消費者在消費消息時是競爭的關系，先得到的先消費，一旦消費完成，該消息就會出隊列，
就不能被其他消費者再消費了，即“一次性消費”。
特點：
1.“離線一次性消費”；
離線：指生產者在發送消息時，不需要消費者在線，生產者只需要將消息發送到MQ隊列中，消費者可以稍后上線取消息；
一次性消費：指消費者之間存在競爭關系，任何一個消費者將消息消費掉之后，其他消費者都不能再進行消費；
若是只有1個生產者和1個消費者，就是我們熟悉的“點對點”通信了；
類似場景對比：
類似送快遞，快遞員（producer）將快遞（Message）放到指定地點（destination）后，就可以離開了，
拿快遞的人（customer）在接收到通知后，到指定地點（destination）去取快遞（Message）就可以了。
當然，取快遞時可能要進行身份驗證，這就涉及到創建連接（connection）時，需要指定用戶名和密碼了。
還有就是，實際生活中，當快遞員把快遞放好之后，照理說應該通知客戶去哪里取快遞，而ActiveMq幫我們
做好了一切，通知的工作Activemq會幫我們實現，而無需我們親自編碼通知消費者，生產者只需要將Message
放到Mq中即可，通知消費者的工作，mq會幫我們處理。
消費者程序：
package com.ll.activemq;import java.util.Collection;import java.util.Iterator;import javax.jms.Connection;import javax.jms.ConnectionFactory;import javax.jms.Destination;import javax.jms.MapMessage;import javax.jms.MessageConsumer;import javax.jms.Session;import javax.jms.TextMessage;import org.apache.activemq.ActiveMQConnection;import org.apache.activemq.ActiveMQConnectionFactory;public class Receiver { public static void main(String[] args) { // ConnectionFactory ：連接工廠，JMS 用它創建連接 ConnectionFactory connectionFactory; // Connection ：JMS 客戶端到JMS Provider 的連接 Connection connection = null; // Session： 一個發送或接收消息的線程 Session session; // Destination ：消息的目的地;消息發送給誰. Destination destination; // 消費者，消息接收者 MessageConsumer consumer; connectionFactory = new ActiveMQConnectionFactory( ActiveMQConnection.DEFAULT_USER, ActiveMQConnection.DEFAULT_PASSWORD, "tcp://localhost:61616"); try { // 構造從工廠得到連接對象 connection = connectionFactory.createConnection(); // 啟動 connection.start(); // 獲取操作連接 session = connection.createSession(Boolean.FALSE, Session.AUTO_ACKNOWLEDGE); // 獲取session注意參數值xingbo.xu-queue是一個服務器的queue，須在在ActiveMq的console配置 destination = session.createQueue("FirstQueue"); consumer = session.createConsumer(destination); // consumer2 = session.createConsumer(destination); System.out.println("消費者1：消費者程序開始運行..."); while (true) { // 設置接收者接收消息的時間，為了便於測試，這里誰定為100s TextMessage message = (TextMessage) consumer.receive(18000); // Object message = (Object) consumer.receive(28000); if (null != message) { System.out.println("消費者1：收到消息-->" + message.getText()); } else { System.out.println("消費者1：運行結束...\n"); break; } } } catch (Exception e) { e.printStackTrace(); } finally { try { if (null != connection) connection.close(); } catch (Throwable ignore) { } } }}
生產者程序：
package com.ll.activemq;import javax.jms.Connection;import javax.jms.ConnectionFactory;import javax.jms.DeliveryMode;import javax.jms.Destination;import javax.jms.MessageProducer;import javax.jms.Session;import javax.jms.TextMessage;import org.apache.activemq.ActiveMQConnection;import org.apache.activemq.ActiveMQConnectionFactory;public class Sender { private static final int SEND_NUMBER = 5; public static void main(String[] args) { // ConnectionFactory ：連接工廠，JMS 用它創建連接 ConnectionFactory connectionFactory; // Connection ：JMS 客戶端到JMS Provider 的連接 Connection connection = null; // Session： 一個發送或接收消息的線程 Session session; // Destination ：消息的目的地;消息發送給誰. Destination destination; // MessageProducer：消息發送者 MessageProducer producer; // TextMessage message; // 構造ConnectionFactory實例對象，此處采用ActiveMq的實現jar connectionFactory = new ActiveMQConnectionFactory( ActiveMQConnection.DEFAULT_USER, ActiveMQConnection.DEFAULT_PASSWORD, "tcp://localhost:61616"); try { // 構造從工廠得到連接對象 connection = connectionFactory.createConnection(); // 啟動 connection.start(); // 獲取操作連接 session = connection.createSession(Boolean.TRUE, Session.AUTO_ACKNOWLEDGE); // 獲取session注意參數值xingbo.xu-queue是一個服務器的queue，須在在ActiveMq的console配置 destination = session.createQueue("FirstQueue"); // 得到消息生成者【發送者】 producer = session.createProducer(destination); // 設置不持久化，此處學習，實際根據項目決定 producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT); // 構造消息，此處寫死，項目就是參數，或者方法獲取 sendMessage(session, producer); session.commit(); } catch (Exception e) { e.printStackTrace(); } finally { try { if (null != connection) connection.close(); } catch (Throwable ignore) { } } } public static void sendMessage(Session session, MessageProducer producer) throws Exception { for (int i = 1; i <= SEND_NUMBER; i++) { TextMessage message = session .createTextMessage("ActiveMq 發送的消息" + i); // 發送消息到目的地方 System.out.println("發送消息：" + "ActiveMq 發送的消息" + i); producer.send(message); } System.out.println("生產者程序退出..."); }}
附件列表