ActiveMQ 3. Java编码实现ActiveMQ通讯
3.1. 队列（Queue）
目的地（Destination）分为：
- 点对点的队列（Queue）
- 一对多的主题（Topic）
3.1.1. 上手代码
pom.xml
org.apache.activemq activemq-all 5.15.9 org.apache.xbean xbean-spring 3.16 生产者代码
public class JmsProducer {
public static final String ACTIVEMQ_URL = "tcp://192.168.181.128:61616/"; public static final String QUEUE_NAME = "queue01"; public static void main(String[] args) throws JMSException { // 1. 创建连接工厂，按照给定的URL地址，采用默认用户名密码 ActiveMQConnectionFactory activeMQConnectionFactory = new ActiveMQConnectionFactory(ACTIVEMQ_URL); // 2. 通过连接工厂，获得连接Connection并启动访问 Connection connection = activeMQConnectionFactory.createConnection(); connection.start(); // 3. 创建会话Session // 两个参数，第一个是事务控制，第二个是签收控制 Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); // 4. 创建目的地（具体是队列queue或主题topic） Queue queue = session.createQueue(QUEUE_NAME); // 5. 创建消息的生产者 MessageProducer messageProducer = session.createProducer(queue); // 6. 通过消息生产者发送消息 for (int i = 0; i < 3; i++) { // 7. 创建消息 TextMessage textMessage = session.createTextMessage("msg---" + i); // 8. 发送给MQ messageProducer.send(textMessage); } // 9. 关闭资源 messageProducer.close(); session.close(); connection.close(); System.out.println("*****消息发布到MQ完成*****"); }}
消费者代码
public class JmsConsumer {
public static final String ACTIVEMQ_URL = "tcp://192.168.181.128:61616/"; public static final String QUEUE_NAME = "queue01"; public static void main(String[] args) throws JMSException { // 1. 创建连接工厂，按照给定的URL地址，采用默认用户名密码 ActiveMQConnectionFactory activeMQConnectionFactory = new ActiveMQConnectionFactory(ACTIVEMQ_URL); // 2. 通过连接工厂，获得连接Connection并启动访问 Connection connection = activeMQConnectionFactory.createConnection(); connection.start(); // 3. 创建会话Session // 两个参数，第一个是事务控制，第二个是签收控制 Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); // 4. 创建目的地（具体是队列queue或主题topic） Queue queue = session.createQueue(QUEUE_NAME); // 5. 创建消费者 MessageConsumer messageConsumer = session.createConsumer(queue); while (true) { TextMessage textMessage = (TextMessage) messageConsumer.receive(); if (textMessage != null) { System.out.println("*****消费者收到消息：" + textMessage.getText()); } else { break; } } // 6. 关闭资源 messageConsumer.close(); session.close(); connection.close(); }}
3.1.2. receive()方法说明
// 收到消息前一直阻塞进程
javax.jms.MessageConsumer#receive()
// 超时后不再阻塞进程
javax.jms.MessageConsumer#receive(long timeout)
3.1.3. 消费者监听器方式接收消息
监听器方式属于异步非阻塞方式，所以需要手动阻塞进程
messageConsumer.setMessageListener(new MessageListener() {
@SneakyThrows
@Override
public void onMessage(Message message) {
if (null != message && message instanceof TextMessage) {
System.out.println("消费者监听器监听到消息***********" + ((TextMessage) message).getText());
}
}
});
// 手动阻塞进程
System.in.read();
3.1.4. 消费者三大消费情况
先生产，只启动1号消费者。问题：1号消费者可以消费消息吗？
可以
先生产，先启动1号消费者，再启动2号消费者。问题：2号消费者可以消费消息吗？
1号消费者可以消费消息；2号消费者不可以消费消息；
先启动2个消费者，再生产6条消息。问题：消费情况如何？
2个消费者各消费一半消息；
3.1.5. 两种消费方式
同步阻塞方式（
receive()）异步非阻塞方式（消费者监听器
onMessage()）
3.1.6. 点对点消息传递域的特点
每个消息只能有一个消费者，类似1对1的关系，类似于快递；
消息的消费者和生产者没有时间上的相关性，类似于短信；
消息被消费后队列中不会再存储，所以消费者不会消费到已经被消费掉的消息；
3.2. 主题（Topic）
3.2.1. 发布订阅消息传递域的特点
- 每个消息可以有多个消费者，属于一对多的关系；
- 生产者和消费者有时间上的相关性，订阅一个主题的消费者只能消费自它订阅之后发布的消息；
- 生产者生产时，topic不保存消息，它是无状态的不落地，假如无人订阅就去生产，那就是一条废消息，所以，一般先启动消费者再启动生产者；
JMS规范允许客户创建持久订阅，这在一定程度上放松了时间上的相关性要求。持久订阅允许消费者消费它在未处于激活状态时的消息。一句话，类似微信公众号订阅。
3.2.2. 上手代码
测试时要先启动消费者，后启动生产者。
生产者代码
public class JmsProducer_Topic {
public static final String ACTIVEMQ_URL = "tcp://192.168.181.128:61616/"; public static final String TOPIC_NAME = "topic01"; public static void main(String[] args) throws JMSException { // 1. 创建连接工厂，按照给定的URL地址，采用默认用户名密码 ActiveMQConnectionFactory activeMQConnectionFactory = new ActiveMQConnectionFactory(ACTIVEMQ_URL); // 2. 通过连接工厂，获得连接Connection并启动访问 Connection connection = activeMQConnectionFactory.createConnection(); connection.start(); // 3. 创建会话Session // 两个参数，第一个是事务控制，第二个是签收控制 Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); // 4. 创建目的地（具体是队列queue或主题topic） Topic topic = session.createTopic(TOPIC_NAME); // 5. 创建消息的生产者 MessageProducer messageProducer = session.createProducer(topic); // 6. 通过消息生产者发送消息 for (int i = 0; i < 3; i++) { // 7. 创建消息 TextMessage textMessage = session.createTextMessage("topic---" + i); // 8. 发送给MQ messageProducer.send(textMessage); } // 9. 关闭资源 messageProducer.close(); session.close(); connection.close(); System.out.println("*****topic消息发布到MQ完成*****"); }}
消费者代码
public class JmsConsumer_Topic {
public static final String ACTIVEMQ_URL = "tcp://192.168.181.128:61616/"; public static final String TOPIC_NAME = "topic01"; public static void main(String[] args) throws JMSException, IOException { System.out.println("我是1号消费者"); // System.out.println("我是2号消费者"); // System.out.println("我是3号消费者"); // 1. 创建连接工厂，按照给定的URL地址，采用默认用户名密码 ActiveMQConnectionFactory activeMQConnectionFactory = new ActiveMQConnectionFactory(ACTIVEMQ_URL); // 2. 通过连接工厂，获得连接Connection并启动访问 Connection connection = activeMQConnectionFactory.createConnection(); connection.start(); // 3. 创建会话Session // 两个参数，第一个是事务控制，第二个是签收控制 Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); // 4. 创建目的地（具体是队列queue或主题topic） Topic topic = session.createTopic(TOPIC_NAME); // 5. 创建消费者 MessageConsumer messageConsumer = session.createConsumer(topic); messageConsumer.setMessageListener(new MessageListener() { @SneakyThrows @Override public void onMessage(Message message) { if (null != message && message instanceof TextMessage) { System.out.println("消费者监听器监听到 TOPIC 消息***********" + ((TextMessage) message).getText()); } } }); // 手动阻塞进程 System.in.read(); // 6. 关闭资源 messageConsumer.close(); session.close(); connection.close(); }}
3.3. 两种模式比较
比较项目
Topic 模式
Queue模式
工作模式
“订阅-发布”模式，如果当前没有订阅者，消息将会被丢弃；如果有多个订阅者，那么这些订阅者都会收到消息
"负载均衡"模式，如果当前没有消费者，消息也不会丢弃；如果有多个消费者，那么一条消息只会发送给其中一个消费者，并且要求消费者ack消息
有无状态
无状态
Queue数据默认会在MQ服务器上以文件形式保存。也可以配置成DB存储
传递完整性
如果没有订阅者，消息会被丢弃
消息不会被丢弃
处理效率
由于消息要按照订阅者数量进行复制，所以处理性能会随着订阅者的增加而明显降低，并且还要结合不同消息协议自身的性能差异
由于一条消息只发送给一个消费者，所以就算消费者再多，性能也不会明显降低。当然不同消息协议的具体性能也是有差异的
