在上一篇《ActiveMQ入门系列二:入门代码实例(点对点模式)》中提到了ActiveMQ中的两种模式:点对点模式(PTP)和发布/订阅模式(Pub & Sub),详细介绍了点对点模式并用代码实例进行说明,今天就介绍下发布/订阅模式。
一、理论基础
发布/订阅模式的工作示意图:
- 消息生产者将消息(发布)到topic中,可以同时有多个消息消费者(订阅)消费该消息。
- 和点对点方式不同,发布到topic的消息会被所有订阅者消费。
- 当生产者发布消息,不管是否有消费者,都不会保存消息。
- 一定要先有消息的消费者,后有消息的生产者。
二、代码实现
- 生产者
package com.sam.topic; import org.apache.activemq.ActiveMQConnectionFactory; import javax.jms.*; /**
* @author JAVA开发老菜鸟
*
*/
public class TopicProducer { public static final String QUEUE_NAME = "topic-demo";//队列名 public void producer(String message) throws JMSException {
ConnectionFactory factory = null;
Connection connection = null;
Session session = null;
MessageProducer producer = null;
try {
/**
* 1.创建连接工厂
* 创建工厂,构造方法有三个参数:分别是用户名、密码、连接地址
* 无参构造:有默认的连接地址,localhost
* 一个参数:无验证模式,无用户的认证
* 三个参数:有认证和连接地址
*/
factory = new ActiveMQConnectionFactory("admin","admin","tcp://localhost:61616");
/**
* 2.创建连接
* 无参数
* 有参数:用户名、密码
*/
connection = factory.createConnection();
/**
* 3.启动连接
* 生产者可以不启动,因为在发送消息的时候回进行检查
* 如果未启动连接,会自动启动
* 如果有特殊配置,需要配置完成后再启动连接
*/
connection.start();
/**
* 4.用连接创建会话
* 有两个参数:是否需要事务、消息确认机制
* 如果支持事务,对于生产者来说第二个参数就无效了,建议传入Session.SESSION_TRANSACTED
* 如果不支持事务,第二个参数必须传递且有效
*
* AUTO_ACKNOWLEDGE:自动确认,消息处理后自动确认(商业开发不推荐)
* CLIENT_ACKNOWLEDGE:客户端手动确认,消费者处理后必须手动确认
* DUPS_OK_ACKNOWLEDGE:有副本的客户端手动确认,消息可以多次处理(不建议)
*/
session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
/**
* 5.用会话创建目的地(主题)、生产者、消息
* 队列名是队列的唯一标记
* 创建生产者的时候可以不指定目的地,可以在发送的时候指定
*/
Destination destination = session.createTopic(QUEUE_NAME);
producer = session.createProducer(destination);
TextMessage textMessage = session.createTextMessage(message);
/**
* 6.生产者发送消息到目的地
*/
producer.send(textMessage);
System.out.println("消息发送成功");
} catch(Exception ex){
throw ex;
} finally {
/**
* 7.释放资源
*/
if(producer != null){
producer.close();
} if(session != null){
session.close();
} if(connection != null){
connection.close();
}
}
} public static void main(String[] args){
TopicProducer producer = new TopicProducer();
try{
producer.producer("hello, activemq");
} catch (Exception ex){
ex.printStackTrace();
}
}
}发布/订阅模式的生产者和点对点模式的代码主要区别就是Destination的创建方式,点对点模式是调用session.createQueue(QUEUE_NAME),而发布/订阅模式是调用session.createTopic(QUEUE_NAME)。
- 消费者
package com.sam.topic; import org.apache.activemq.ActiveMQConnectionFactory; import javax.jms.*;
import java.io.IOException; /**
* @author JAVA开发老菜鸟
*
* 观察者消费--监听消费
*/
public class TopicConsumer { public void consumer() throws JMSException, IOException {
ConnectionFactory factory = null;
Connection connection = null;
Session session = null;
MessageConsumer consumer = null;
try {
factory = new ActiveMQConnectionFactory("admin","admin","tcp://localhost:61616");
connection = factory.createConnection();
/**
* 消费者必须启动连接,否则无法消费
*/
connection.start();
session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
Destination destination = session.createTopic(TopicProducer.QUEUE_NAME);
consumer = session.createConsumer(destination);
/**
* 注册监听器,队列中的消息变化会自动触发监听器,接收并自动处理消息
*
* 监听器一旦注册,永久有效,一直到程序关闭
* 监听器可以注册多个,相当于集群
* activemq自动轮询多个监听器,实现并行处理
*/
consumer.setMessageListener(new MessageListener() {
@Override
public void onMessage(Message message) { try {
TextMessage om = (TextMessage) message;
String data = om.getText();
System.out.println(data);
} catch (JMSException e) {
e.printStackTrace();
}
}
});
} catch(Exception ex){
throw ex;
}
} public static void main(String[] args){
TopicConsumer consumer = new TopicConsumer();
try{
consumer.consumer();
} catch (Exception ex){
ex.printStackTrace();
}
}
}消费者在点对点监听消费的基础上进行变化,主要区别有两个:1.同生产者一样,也是Destination的创建方式不同; 2.消息无需手动确认,直接采用自动确认机制
代码写完了,接下来进行测试,由于subscribe可以有多个,而且每个都可以消费到相同的消息,因此我们消费者启动两个。
先执行生产者
在控制台页面的Topics下出现了我定义的topic并且有1条消息发送成功且未消费
然后执行两个消费者,两个消费者都没有消费到任何消息
并且,控制台页面只是多了2个消费者,已经消费的消息还是0
为什么呢?还记得前面的理论基础说的吗?就是这个原因
继续,我们在两个消费者启动好的前提下,再执行生产者, 这个时候会发现两个消费者都消费了该消息
再看下控制台页面
已消费消息这里是2,这个2并不是说之前发的两个消息都消费了,而是说第二个消息消费了2次, 1 * 2 = 2
不信的话,可以再执行一遍生产者,这个时候就是4,而不是3
累计发送过3条消息,消息消费了4次,这里的4就是后面两条分别被消费了2次, 2 * 2 = 4
三、两种模式比较
好,到这里,发布/订阅模式就介绍完了。
如果有收获,就点个赞呗