在官網(wǎng)下載ActiveMQ安裝包后，打開安裝目錄下的/bin/win64/InstallService.bat文件即可安裝ActiveMQ服務(wù)?？纱蜷_服務(wù)面板，啟動(dòng)ActiveMQ服務(wù)。
瀏覽器打開http://localhost:8161/頁面，即可進(jìn)入ActiveMQ管理頁面，初始賬號(hào)密碼均為admin
一、隊(duì)列模式
（一）概念
隊(duì)列模式又稱點(diǎn)對(duì)點(diǎn)模式（Point-to-Point Messaging Domain）。
在點(diǎn)對(duì)點(diǎn)通信模式中，應(yīng)用程序由消息隊(duì)列，發(fā)送方，接收方組成。每個(gè)消息都被發(fā)送到一個(gè)特定的隊(duì)列，接收者從隊(duì)列中獲取消息。隊(duì)列保留著消息，直到他們被消費(fèi)或超時(shí)。
特點(diǎn)：
每個(gè)消息只要一個(gè)消費(fèi)者
發(fā)送者和接收者在時(shí)間上是沒有時(shí)間的約束，也就是說發(fā)送者在發(fā)送完消息之后，不管接收者有沒有接受消息，都不會(huì)影響發(fā)送方發(fā)送消息到消息隊(duì)列中。
發(fā)送方不管是否在發(fā)送消息，接收方都可以從消息隊(duì)列中去到消息
-
接收方在接收完消息之后，需要向消息隊(duì)列應(yīng)答成功
（二）代碼實(shí)現(xiàn)
生產(chǎn)者：
import org.apache.activemq.ActiveMQConnectionFactory;
import javax.jms.*;
public class AppProducer {
private static final String url = "tcp://127.0.0.1:61616";
private static final String queueName="queue-test";
public static void main(String[] args) throws JMSException {
ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(url);
Connection connection = connectionFactory.createConnection();
connection.start();
Session session = connection.createSession(false,Session.AUTO_ACKNOWLEDGE);
//創(chuàng)建隊(duì)列目標(biāo)
Destination destination = session.createQueue(queueName);
MessageProducer producer = session.createProducer(destination);
for (int i = 0; i < 100; i++) {
TextMessage message = session.createTextMessage("message"+i);
producer.send(message);
System.out.println("發(fā)送消息:"+message.getText());
}
connection.close();
}
}
消費(fèi)者：
import org.apache.activemq.ActiveMQConnectionFactory;
import javax.jms.*;
public class AppConsumer {
private static final String url = "tcp://127.0.0.1:61616";
private static final String queueName="queue-test";
public static void main(String[] args) throws JMSException {
ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(url);
Connection connection = connectionFactory.createConnection();
connection.start();
Session session = connection.createSession(false,Session.AUTO_ACKNOWLEDGE);
Destination destination = session.createQueue(queueName);
MessageConsumer consumer = session.createConsumer(destination);
consumer.setMessageListener(new MessageListener() {
public void onMessage(Message message) {
TextMessage textMessage = (TextMessage) message;
try {
System.out.println("接收消息:"+textMessage.getText());
} catch (JMSException e) {
e.printStackTrace();
}
}
});
//connection.close();
}
}
二、主題模式
（一）概念
主題模式又稱發(fā)布/訂閱模式（Publish/Subscribe Messaging Domain ）。
在發(fā)布/訂閱消息模型中，發(fā)布者發(fā)布一個(gè)消息，該消息通過topic傳遞給所有的客戶端。該模式下，發(fā)布者與訂閱者都是匿名的，即發(fā)布者與訂閱者都不知道對(duì)方是誰。并且可以動(dòng)態(tài)的發(fā)布與訂閱Topic。Topic主要用于保存和傳遞消息，且會(huì)一直保存消息直到消息被傳遞給客戶端。
特點(diǎn)：
一個(gè)消息可以傳遞個(gè)多個(gè)訂閱者（即：一個(gè)消息可以有多個(gè)接受方）
發(fā)布者與訂閱者具有時(shí)間約束，針對(duì)某個(gè)主題（Topic）的訂閱者，它必須創(chuàng)建一個(gè)訂閱者之后，才能消費(fèi)發(fā)布者的消息，而且為了消費(fèi)消息，訂閱者必須保持運(yùn)行的狀態(tài)。
-
為了緩和這樣嚴(yán)格的時(shí)間相關(guān)性，JMS允許訂閱者創(chuàng)建一個(gè)可持久化的訂閱。這樣，即使訂閱者沒有被激活（運(yùn)行），它也能接收到發(fā)布者的消息。
（二）代碼實(shí)現(xiàn)
生產(chǎn)者：
import org.apache.activemq.ActiveMQConnectionFactory;
import javax.jms.*;
public class AppProducer {
private static final String url = "tcp://127.0.0.1:61616";
private static final String topicName="topic-test";
public static void main(String[] args) throws JMSException {
ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(url);
Connection connection = connectionFactory.createConnection();
connection.start();
Session session = connection.createSession(false,Session.AUTO_ACKNOWLEDGE);
//創(chuàng)建主題目標(biāo)
Destination destination = session.createTopic(topicName);
MessageProducer producer = session.createProducer(destination);
for (int i = 0; i < 100; i++) {
TextMessage message = session.createTextMessage("message"+i);
producer.send(message);
System.out.println("發(fā)送消息:"+message.getText());
}
connection.close();
}
}
消費(fèi)者：
import org.apache.activemq.ActiveMQConnectionFactory;
import javax.jms.*;
public class AppConsumer {
private static final String url = "tcp://127.0.0.1:61616";
private static final String topicName="topic-test";
public static void main(String[] args) throws JMSException {
ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(url);
Connection connection = connectionFactory.createConnection();
connection.start();
Session session = connection.createSession(false,Session.AUTO_ACKNOWLEDGE);
Destination destination = session.createTopic(topicName);
MessageConsumer consumer = session.createConsumer(destination);
consumer.setMessageListener(new MessageListener() {
public void onMessage(Message message) {
TextMessage textMessage = (TextMessage) message;
try {
System.out.println("接收消息:"+textMessage.getText());
} catch (JMSException e) {
e.printStackTrace();
}
}
});
//connection.close();
}
}
- 進(jìn)入127.0.0.1:8161/admin頁面，點(diǎn)擊Queue可查看隊(duì)列消息，點(diǎn)擊Topic可查看主題消息。
- 【兩種方式代碼差別不大，僅僅修改目標(biāo)對(duì)象即可?！?/li>
//創(chuàng)建隊(duì)列目標(biāo)
Destination destination = session.createQueue(queueName);
//創(chuàng)建主題目標(biāo)
Destination destination = session.createTopic(topicName);
