在官网下载ActiveMQ安装包后，打开安装目录下的/bin/win64/InstallService.bat文件即可安装ActiveMQ服务。可打开服务面板，启动ActiveMQ服务。
浏览器打开http://localhost:8161/页面，即可进入ActiveMQ管理页面，初始账号密码均为admin
一、队列模式
（一）概念
队列模式又称点对点模式（Point-to-Point Messaging Domain）。
在点对点通信模式中，应用程序由消息队列，发送方，接收方组成。每个消息都被发送到一个特定的队列，接收者从队列中获取消息。队列保留着消息，直到他们被消费或超时。
特点：
-
每个消息只要一个消费者
-
发送者和接收者在时间上是没有时间的约束，也就是说发送者在发送完消息之后，不管接收者有没有接受消息，都不会影响发送方发送消息到消息队列中。
-
发送方不管是否在发送消息，接收方都可以从消息队列中去到消息
-
接收方在接收完消息之后，需要向消息队列应答成功
（二）代码实现
生产者：
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);
//创建队列目标
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("发送消息:"+message.getText());
}
connection.close();
}
}
消费者：
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();
}
}
二、主题模式
（一）概念
主题模式又称发布/订阅模式（Publish/Subscribe Messaging Domain ）。
在发布/订阅消息模型中，发布者发布一个消息，该消息通过topic传递给所有的客户端。该模式下，发布者与订阅者都是匿名的，即发布者与订阅者都不知道对方是谁。并且可以动态的发布与订阅Topic。Topic主要用于保存和传递消息，且会一直保存消息直到消息被传递给客户端。
特点：
-
一个消息可以传递个多个订阅者（即：一个消息可以有多个接受方）
-
发布者与订阅者具有时间约束，针对某个主题（Topic）的订阅者，它必须创建一个订阅者之后，才能消费发布者的消息，而且为了消费消息，订阅者必须保持运行的状态。
-
为了缓和这样严格的时间相关性，JMS允许订阅者创建一个可持久化的订阅。这样，即使订阅者没有被激活（运行），它也能接收到发布者的消息。
（二）代码实现
生产者：
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);
//创建主题目标
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("发送消息:"+message.getText());
}
connection.close();
}
}
消费者：
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();
}
}
- 进入127.0.0.1:8161/admin页面，点击Queue可查看队列消息，点击Topic可查看主题消息。
- 【两种方式代码差别不大，仅仅修改目标对象即可。】
//创建队列目标
Destination destination = session.createQueue(queueName);
//创建主题目标
Destination destination = session.createTopic(topicName);
网友评论