ActiveMQ 是Apache出品,最流行的,能力强劲的开源消息总线。ActiveMQ 是一个彻底支持JMS1.1和J2EE 1.4规范的 JMS Provider实现,尽管JMS规范出台已是好久的事情了,可是JMS在当今的J2EE应用中间仍然扮演着特殊的地位。前端
在介绍activemq以前,先简单介绍JMS,它是J2EE的13个规范之一,提供的是消息中间件的规范。
1、JMS包括如下基本构件:java
- 链接工厂,
是客户用来建立链接的对象,ActiveMQ提供的是ActiveMQConnectionFactory;- 链接connection;
- 会话session,是发送和接收消息的上下文,用于建立消息生产者,消息消费者,相比rocketMQ会话session是提供事务性的;
- 目的地destination,指定生产消息的目的地和消费消息的来源对象;
- 生产者,由会话建立的对象。
- 消费者,由会话建立的对象。
2、消息通讯机制
点对点模式,每一个消息只有1个消费者,它的目的地称为queue队列;
发布/订阅模式,每一个消息能够有多个消费者,并且订阅一个主题的消费者,只能消费自它订阅以后发布的消息。
消息确认机制
Session.AUTO_ACKNOWLEDGE,直接使用receive方法。Session.CLIENT_ACKNOWLEDGE,经过消息的acknowledge 方法确认消息。Session.DUPS_ACKNOWLEDGE,该选择只是会话迟钝第确认消息的提交。若是JMS provider 失败,那么可能会致使一些重复的消息。若是是重复的消息,那么JMS provider 必须把消息头的JMSRedelivered 字段设置为true。web
⒈ 多种语言和协议编写客户端。语言: Java,C,C++,C#,Ruby,Perl,Python,PHP。应用协议: OpenWire,Stomp REST,WS Notification,XMPP,AMQP
⒉ 彻底支持JMS1.1和J2EE 1.4规范 (持久化,XA消息,事务)
⒊ 对Spring的支持,ActiveMQ能够很容易内嵌到使用Spring的系统里面去,并且也支持Spring2.0的特性
⒋ 经过了常见J2EE服务器(如 Geronimo,JBoss 4,GlassFish,WebLogic)的测试,其中经过JCA 1.5 resource adaptors的配置,可让ActiveMQ能够自动的部署到任何兼容J2EE 1.4 商业服务器上
⒌ 支持多种传送协议:in-VM,TCP,SSL,NIO,UDP,JGroups,JXTA
⒍ 支持经过JDBC和journal提供高速的消息持久化
⒎ 从设计上保证了高性能的集群,客户端-服务器,点对点
⒏ 支持Ajax
⒐ 支持与Axis的整合
⒑ 能够很容易的调用内嵌JMS provider,进行测试数据库
参考我之前文章MQ(message queue)使用 Spring整合 MQ下载 五分钟上手使用 注:文档连接为SpringBoot整合ActiveMQ教程 文章开头为安装教程附下载apache
使用JMS原生API编写测试类,向消息中间件写入消息的开发步骤:服务器
1 建立链接工厂
2 建立链接
3 启动链接
4 创建会话
5 建立队列
6 建立生产者
7 建立消息
8 发送消息
9 提交网络
<dependencies> <dependency> <groupId>org.apache.activemq</groupId> <artifactId>activemq-all</artifactId> <version>5.14.0</version> </dependency> </dependencies>
import org.apache.activemq.ActiveMQConnectionFactory; import javax.jms.*; public class ActiveMQProducter { public static void main(String[] args) throws Exception{ // 链接工厂 // 使用默认用户名、密码、路径 // 由于:底层实现:final String defaultURL = "tcp://" + DEFAULT_BROKER_HOST + ":" + DEFAULT_BROKER_PORT; // 因此:路径 tcp://host:61616 //1 建立链接工厂 ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory(); //2 建立链接 Connection connection = connectionFactory.createConnection(); //3 打开链接 connection.start(); //4 建立会话 //第一个参数:是否开启事务 //第二个参数:消息是否自动确认 Session session = connection.createSession(true, Session.AUTO_ACKNOWLEDGE); //建立队列 Queue queue = session.createQueue("hello20181119"); //5 建立生产者 MessageProducer producer = session.createProducer(queue); //6 建立消息 Message message = session.createTextMessage("helloworld"); //7 发送消息 producer.send(message); //8 关闭消息 session.commit(); producer.close(); session.close(); connection.close(); System.out.println("消息生产成功"); } }
运行后结果 默认tcp链接activeMQ端口 61616 !!! 程序链接端口
查看网页 网页使用端口8161访问session
访问:http://localhost:8161/ 点击:Manage ActiveMQ broker
用户名和密码 都是 admin架构
消费前:表示没有消费 有一个消息在队列中
并发
import org.apache.activemq.ActiveMQConnectionFactory; import javax.jms.*; public class ActiveMQConsumer { public static void main(String[] args) throws Exception { //建立链接工厂 ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory(); //建立链接 Connection connection = connectionFactory.createConnection(); //开启链接 connection.start(); //建立会话 /** 第一个参数,是否使用事务 若是设置true,操做消息队列后,必须使用 session.commit(); 若是设置false,操做消息队列后,不使用session.commit(); */ Session session = connection.createSession(true, Session.AUTO_ACKNOWLEDGE); //建立队列 Queue queue = session.createQueue("hello20181119"); //建立消费者 MessageConsumer consumer = session.createConsumer(queue); while(true){ //失效时间,若是10秒内没有收到新的消息,说明没有消息存在,此时能够退出当前循环 TextMessage message = (TextMessage) consumer.receive(10000); if(message!=null){ System.out.println(message.getText()); }else { break; } } //关闭链接 session.commit(); session.close(); connection.close(); System.out.println("消费结束0"); } }
注意:测试的时候必定先启动消费者,而后再启动生产者(Queue队列)
查看控制台,发现信息已经被消费
查看网页
// 使用监听器消费 public static void main(String[] args) throws Exception { // 链接工厂 // 使用默认用户名、密码、路径 // 路径 tcp://host:61616 ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(); // 获取一个链接 Connection connection = connectionFactory.createConnection(); // 开启链接 connection.start(); // 创建会话 // 第一个参数,是否使用事务,若是设置true,操做消息队列后,必须使用 session.commit(); Session session = connection.createSession(false,Session.AUTO_ACKNOWLEDGE); // 建立队列或者话题对象 Queue queue = session.createQueue("hello20181119"); // 建立消费者 MessageConsumer messageConsumer = session.createConsumer(queue); messageConsumer.setMessageListener(new MessageListener() { // 每次接收消息,自动调用 onMessage public void onMessage(Message message) { TextMessage textMessage = (TextMessage) message; try { System.out.println(textMessage.getText()); } catch (JMSException e) { e.printStackTrace(); } } }); //此时,不能让程序结束,若是结束,监听就结束了 while (true) { // 目的:不能让程序死掉 } }
如下介绍消息队列在实际应用中经常使用的使用场景。
异步处理,应用解耦,流量削锋和消息通信四个场景
场景说明:用户注册后,须要发注册邮件和注册短信。传统的作法有两种:
1.串行的方式;2.并行方式
(1)串行方式:将注册信息写入数据库
成功后,发送注册邮件,再发送注册短信。以上三个任务所有完成后,返回给客户端
(2)并行方式:将注册信息写入数据库成功后,发送注册邮件的同时,发送注册短信。以上三个任务完成后,返回给客户端。与串行的差异是,并行的方式能够提升处理的时间
假设三个业务节点每一个使用50毫秒钟,不考虑网络等其余开销,则串行方式的时间是150毫秒,并行的时间多是100毫秒。
由于CPU在单位时间内处理的请求数是必定的,假设CPU在1秒内吞吐量是100次。则串行方式1秒内CPU可处理的请求量是7次(1000/150)。并行方式处理的请求量是10次(1000/100)
小结:如以上案例描述,传统的方式系统的性能(并发量,吞吐量,响应时间)会有瓶颈。如何解决这个问题呢?
引入消息队列,将不是必须的业务逻辑,异步处理。改造后的架构以下:
按照以上约定,用户的响应时间至关因而注册信息写入数据库的时间,也就是50毫秒。注册邮件,发送短信写入消息队列后,直接返回,所以写入消息队列的速度很快,基本能够忽略,所以用户的响应时间多是50毫秒。所以架构改变后,系统的吞吐量提升到每秒20 QPS。比串行提升了3倍,比并行提升了2倍
场景说明:用户下单后,订单系统须要通知库存系统。传统的作法是,订单系统调用库存系统的接口。以下图 以我作的项目举例
传统模式的缺点:
l 假如库存系统没法访问,则订单减库存将失败,从而致使订单失败
l 订单系统与库存系统耦合
如何解决以上问题呢?引入应用消息队列后的方案,以下图:
订单系统:用户下单后,订单系统完成持久化处理,将消息写入消息队列,返回用户订单下单成功
库存系统:订阅下单的消息,采用pub/sub(发布/订阅)的方式,获取下单信息,库存系统根据下单信息,进行库存操做
假如:在下单时库存系统不能正常使用。也不影响正常下单,由于下单后,订单系统写入消息队列就再也不关心其余的后续操做了。实现订单系统与库存系统的应用解耦
流量削锋也是消息队列中的经常使用场景,通常在秒杀或团抢活动中使用普遍
应用场景:秒杀活动,通常会由于流量过大,致使流量暴增,应用挂掉。为解决这个问题,通常须要在应用前端加入消息队列。
日志处理是指将消息队列用在日志处理中,好比Kafka的应用,解决大量日志传输的问题。架构简化以下
(2)Logstash:作日志解析,统一成JSON输出给Elasticsearch
(3)Elasticsearch:实时日志分析服务的核心技术,一个schemaless,实时的数据存储服务,经过index组织数据,兼具强大的搜索和统计功能
(4)Kibana:基于Elasticsearch的数据可视化组件,超强的数据可视化能力是众多公司选择ELK stack的重要缘由
消息通信是指,消息队列通常都内置了高效的通讯机制,所以也能够用在纯的消息通信。好比实现点对点消息队列,或者聊天室等
点对点通信:
客户端A和客户端B使用同一队列,进行消息通信。
聊天室通信:
客户端A,客户端B,客户端N订阅同一主题,进行消息发布和接收。实现相似聊天室效果。
以上实际是消息队列的两种消息模式,点对点或发布订阅模式。模型为示意图,供参考。
消息队列的JAVAEE规范JMS 。JMS(Java Message Service,java消息服务)API是一个消息服务的标准/规范,容许应用程序组件基于JavaEE平台建立、发送、接收和读取消息。它使分布式通讯耦合度更低,消息服务更加可靠以及异步性。
在JMS标准中,有两种消息模型P2P(Point to Point),Publish/Subscribe(Pub/Sub)。
P2P模式包含三个角色:消息队列(Queue),发送者(Sender),接收者(Receiver)。每一个消息都被发送到一个特定的队列,接收者从队列中获取消息。队列保留着消息,直到他们被消费或超时。
P2P的特色
l 每一个消息只能被一个消费者(Consumer)消费(即一旦被消费,消息就再也不存在于消息队列中)
l 发送者和接收者之间在时间上没有依赖性,也就是说当发送者发送了消息以后,无论接收者有没有正在运行,它不会影响到消息被发送到队列
l 接收者在成功接收消息以后需向队列应答成功
若是但愿发送的每一个消息都会被成功处理的话,那么须要P2P模式。
包含三个角色主题(Topic),发布者(Publisher),订阅者(Subscriber) 多个发布者将消息发送到Topic,系统将这些消息传递给多个订阅者。
Pub/Sub的特色
为了缓和这样严格的时间相关性,JMS容许订阅者建立一个可持久化的订阅。这样,即便订阅者没有被激活(运行),它也能接收到发布者的消息。
若是但愿发送的消息能够被多个消费者处理的话,那么能够采用Pub/Sub模型。
在JMS中,消息的产生和消费都是异步的。对于消费来讲,JMS的消息者能够经过两种方式来消费消息。
(1)同步
订阅者或接收者经过receive方法来接收消息,receive方法在接收到消息以前(或超时以前)将一直阻塞;
(2)异步
订阅者或接收者能够注册为一个消息监听器。当消息到达以后,系统自动调用监听器的onMessage方法。
祝你幸福
送你一首歌《I Really Like You》Anthem Lights
附图 QQ空间看到的一句话