在上一篇《ActiveMQ入门系列二:入门代码实例(点对点模式)》中提到了ActiveMQ中的两种模式:点对点模式(PTP)和发布/订阅模式(Pub & Sub),详细介绍了点对点模式并用代码实例进行说明,今天就介绍下发布/订阅模式。html
1、理论基础java
发布/订阅模式的工做示意图:apache
2、代码实现session
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)。tcp
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.消息无需手动确认,直接采用自动确认机制ide
代码写完了,接下来进行测试,因为subscribe能够有多个,并且每一个均可以消费到相同的消息,所以咱们消费者启动两个。测试
先执行生产者spa
在控制台页面的Topics下出现了我定义的topic而且有1条消息发送成功且未消费3d
而后执行两个消费者,两个消费者都没有消费到任何消息code
而且,控制台页面只是多了2个消费者,已经消费的消息仍是0
为何呢?还记得前面的理论基础说的吗?就是这个缘由
继续,咱们在两个消费者启动好的前提下,再执行生产者, 这个时候会发现两个消费者都消费了该消息
再看下控制台页面
已消费消息这里是2,这个2并非说以前发的两个消息都消费了,而是说第二个消息消费了2次, 1 * 2 = 2
不信的话,能够再执行一遍生产者,这个时候就是4,而不是3
累计发送过3条消息,消息消费了4次,这里的4就是后面两条分别被消费了2次, 2 * 2 = 4
3、两种模式比较
好,到这里,发布/订阅模式就介绍完了。
若是有收获,就点个赞呗