ActiveMQ的发布—订阅消息模式

一、订阅杂志
我们很多人都订过杂志，其过程很简单。只要告诉邮局我们所要订的杂志名、投递的地址，付了钱就OK。出版社定期会将出版的杂志交给邮局，邮局会根据订阅的列表，将杂志送达消费者手中。这样我们就可以看到每一期精彩的杂志了。

仔细思考一下订杂志的过程，我们会发现这样几个特点：
1、消费者订杂志不需要直接找出版社；
2、出版社只需要把杂志交给邮局；
3、邮局将杂志送达消费者。
邮局在整个过程中扮演了非常重要的中转作用，在出版社和消费者相互不需要知道对方的情况下，邮局完成了杂志的投递。

二、发布-订阅消息模式
刚刚讲了订阅杂志，下面我们会讲传统调用模式演化到发布-订阅消息模式。

有些网站在注册用户成功后发一封激活邮件，用户收到邮件后点击激活链接后才能使用该网站。一般的做法是在注册用户业务逻辑中调用发送邮件的逻辑。这样用户业务就依赖于邮件业务。如果以后改为短信激活，注册用户业务逻辑就必须修改为调用发送短信的逻辑。如果要注册后给用户加点积分，再加一段逻辑。经过多次修改，我们发现很简单的注册用户业务已经越来越复杂，越来越难以维护。相信很多开发者都会有类似痛苦的经历。

即使用户业务实现中对其他业务是接口依赖，也避免不了业务变化带来的依赖影响。怎么办？解耦！将注册用户业务逻辑中注册成功后的处理剥离出来。

再回头看看“订阅杂志”，如果没有邮局，出版社就必须自己将杂志送达所有消费者。这种情形就和现在的注册用户业务一样。我们发现问题了，在用户业务和其他业务之间缺少了邮局所扮角色。

我们把邮局抽象成一个管理消息的地方，叫“消息管理器”。注册用户成功后发送一个消息给消息管理器，由消息管理器转发该消息给需要处理的业务。现在，用户业务只依赖于消息管理器了，它再也不会为了注册用户成功后的其他处理而烦恼。

注册用户的改造就是借鉴了“订阅杂志”这样原始的模式。我们再进一步抽象，用户业务就是消息的“生产者”，它将消息发布到消息管理器。邮件业务就是消息的“消费者”，它将收到的消息进行处理。邮局可以订阅很多种杂志，杂志都是通过某种编号来区分；消息管理器也可以管理多种消息，每种消息都会有一个“主题”来区分，消费者都是通过主题来订阅的。

发布-订阅消息模式已经呈现在我们面前，在这里，对于发布者来说，它和所有的订阅者就构成了一个1对多的关系。这种关系如下图所示：

示例：
1、Publish.java：消息发布者

package com.ljq.durian.test.activemq;import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.Destination;
import javax.jms.MessageProducer;
import javax.jms.Session;
import javax.jms.TextMessage;import org.apache.activemq.ActiveMQConnectionFactory;public class Publish {private ConnectionFactory factory;private Connection connection;private Session session;private MessageProducer producer;public Publish() {try {factory = new ActiveMQConnectionFactory("ljq", "ljq", "failover:(tcp://localhost:61616)?Randomize=false");connection = factory.createConnection();connection.start();session = connection.createSession(Boolean.FALSE, Session.AUTO_ACKNOWLEDGE);producer = session.createProducer(null);} catch (Exception e) {e.printStackTrace();}}public void sendMessage() throws Exception {Destination destination = session.createTopic("Topic001");TextMessage msg = session.createTextMessage("我是消息内容...");producer.send(destination, msg);if(connection != null){connection.close();}    }public static void main(String[] args) throws Exception {Publish publish= new Publish();publish.sendMessage();}
}

2、Subscriber1.java：消息订阅者

package com.ljq.durian.test.activemq;import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageConsumer;
import javax.jms.MessageListener;
import javax.jms.Session;
import javax.jms.TextMessage;import org.apache.activemq.ActiveMQConnectionFactory;public class Subscriber1 {private ConnectionFactory factory;private Connection connection;private Session session;public Subscriber1() {try {factory = new ActiveMQConnectionFactory("ljq", "ljq", "failover:(tcp://localhost:61616)?Randomize=false");connection = factory.createConnection();connection.start();session = connection.createSession(Boolean.FALSE, Session.AUTO_ACKNOWLEDGE);} catch (Exception e) {e.printStackTrace();}}public void receive() throws Exception {Destination topic = session.createTopic("Topic001") ;MessageConsumer consumer = session.createConsumer(topic);consumer.setMessageListener(new Listener());}class Listener implements MessageListener {@Overridepublic void onMessage(Message message) {try {TextMessage tm = (TextMessage) message;System.out.println("Subscriber1 Received message: " + tm.getText());} catch (JMSException e) {e.printStackTrace();}}}public static void main(String[] args) throws Exception {Subscriber1 subscriber = new Subscriber1();subscriber.receive();}
}

3、Subscriber2.java：消息订阅者

package com.ljq.durian.test.activemq;import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageConsumer;
import javax.jms.MessageListener;
import javax.jms.Session;
import javax.jms.TextMessage;import org.apache.activemq.ActiveMQConnectionFactory;public class Subscriber2 {private ConnectionFactory factory;private Connection connection;private Session session;public Subscriber2() {try {factory =new ActiveMQConnectionFactory("ljq", "ljq","failover:(tcp://192.168.1.101:61616)?Randomize=false");connection = factory.createConnection();connection.start();session = connection.createSession(Boolean.FALSE, Session.AUTO_ACKNOWLEDGE);} catch (Exception e) {e.printStackTrace();}}public void receive() throws Exception {Destination topic = session.createTopic("Topic001") ;MessageConsumer consumer = session.createConsumer(topic);consumer.setMessageListener(new Listener());}class Listener implements MessageListener {public void onMessage(Message message) {System.out.println(message);try {TextMessage tm = (TextMessage) message;System.out.println("Subscriber2 Received message: " + tm.getText());} catch (JMSException e) {e.printStackTrace();}}}public static void main(String[] args) throws Exception {Subscriber2 subscriber = new Subscriber2();subscriber.receive();}
}

转自https://www.cnblogs.com/linjiqin/p/6518780.html

ActiveMQ的发布—订阅消息模式相关推荐

ActiveMQ之发布- 订阅消息模式实现
一.概念发布者/订阅者模型支持向一个特定的消息主题发布消息.0或多个订阅者可能对接收来自特定消息主题的消息感兴趣.在这种模型下,发布者和订阅者彼此不知道对方.这种模式好比是匿名公告板.这种模式被概括 ...
java订阅发布模式_Spring Boot ActiveMQ发布/订阅消息模式原理解析
本文在<Spring Boot基于Active MQ实现整合JMS>的基础上,介绍如何使用ActiveMQ的发布/订阅消息模式.发布/订阅消息模式是消息发送者发送消息到主题(topic), ...
ActiveMQ Topic发布订阅消息
2019独角兽企业重金招聘Python工程师标准>>> ActiveMQ Topic发布订阅消息博客分类: MQ package com.googlecode.garbagecan ...
分布式发布订阅消息系统—Apache Kafka
1.什么是Kafka Kafka是一种高吞吐量的分布式发布订阅消息系统,它可以处理消费者规模的网站中的所有动作流数据. 这种动作(网页浏览,搜索和其他用户的行动)是在现代网络上的许多社会功能的一个关键 ...
JavaScript发布订阅者模式
假如你要建立一个网站,通常来说会有许多用户.你作为一名管理者,有时候需要将重要的消息发布给你的用户.在软件开发领域,开发此功能往往用到发布订阅者模式.下面我以简单的javascript来说明. 定义发 ...
aws lambda使用_如何使用AWS Lambda为发布/订阅消息选择最佳事件源
aws lambda使用 by Yan Cui 崔燕如何使用AWS Lambda为发布/订阅消息选择最佳事件源 (How to choose the best event source for pu ...
Spring Boot基础学习笔记25：RabbitMQ - 发布/订阅工作模式
文章目录零.学习目标一.准备工作 (一)创建Spring Boot项目 - PublishSubscribeDemo (二)在应用属性文件里配置RabbitMQ 二.基于API进行消息发布和订阅 ...
kafka 基础知识梳理-kafka是一种高吞吐量的分布式发布订阅消息系统
一.kafka 简介今社会各种应用系统诸如商业.社交.搜索.浏览等像信息工厂一样不断的生产出各种信息,在大数据时代,我们面临如下几个挑战: 如何收集这些巨大的信息如何分析它如何及时做到如上两点 ...
Publish/SubScribe（发布/订阅）模式
前言上一篇文章中,我们讲了如何利用Observer(观察者)模式实现多选框的全选, 本篇文章将带来Publish/SubScribe模式,并且利用该模式实现一个简易的消息通知功能,文章的最后还与Ob ...

ActiveMQ的发布—订阅消息模式

ActiveMQ的发布—订阅消息模式相关推荐

最新文章

热门文章