activemq基础篇1--引入activemq的好处

朱雀 2022-03-17 15:48 220阅读 0赞

> ** 文章最前**： 我是Octopus，这个名字来源于我的中文名--章鱼；我热爱编程、热爱算法、热爱开源。所有源码在我的个人[github][] ；这博客是记录我学习的点点滴滴，如果您对 Python、Java、AI、算法有兴趣，可以关注我的动态，一起学习，共同进步。

** 相关文章：**

1.  [activemq基础篇1--引入activemq的好处][activemq_1--_activemq]
2.  [activemq基础2---源码分析][activemq_2---]
3.  [activemq基础篇3--源码分析2][activemq_3--_2]

--------------------

**文章目录：**

消息中间件的初步认识

消息中间件能做什么？

ActiveMq的简介

MOM的特点：

JMS规范：

jms的体系结构：

消息的传播流程：

jms的基本功能：

消息的结构组成：

持久订阅

在事务行会话中

消息的持久化存储

--------------------

### **消息中间件的初步认识** ###

消息中间件是利用高效可靠的消息传递机制进行平台无关的数据交流，并基于数据通信来进行分布式系统的集成。通过提供消息传递和消息排队模型，可以在分布式架构下扩展进程之间的通信。

### **消息中间件能做什么？** ###

消息中间件主要解决的就是分布式系统之间消息传递的问题，它能够屏蔽各平台以及协议之间的特性，实现应用程序之间的协同。举个例子，一个电商平台的注册功能来简单分析，用户注册一个服务，不单单只是insert一条数据到数据库就完事了，还需要发送激活邮件，发送新人红包或者积分，分送营销短信等一系列操作。假如说这里面的每一个操作 ，都要消耗1S,那么整个注册过程就需要耗时4s才能响应给用户。

![20190225105759824.png][]

但是我们从注册这个服务可以看到，每个子操作都是相对独立的模块，同时，基于领域划分以后，发送激活邮件，发送营销短信，赠送积分以及红包都属于不同的子域。所以我们可以对这些子操作进行实现异步化还行，类似于多线程并行处理的概念。

如何实现异步化？多线程可以实现，只是，消息的持久化，消息的重发这些条件，多线程并不能满足，所需要借助一些开源中间件。

**场合1：实现异步处理；**而分布式消息队列是一个非常好的解决办法，引入分布式消息队列以后，就能大大提升程序的处理效率，并且还解决了各个模块之间的耦合问题。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3p5MzQ1MjkzNzIx_size_16_color_FFFFFF_t_70][]

**                                                                           **异步通信示意图

** 场景2：秒杀场景；**通过分布式队列来实现流量的整型，通过消息队列可以缓解高流量的访问压力。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3p5MzQ1MjkzNzIx_size_16_color_FFFFFF_t_70 1][]

秒杀示意图

1）用户提交过程的请求，先写入到消息队列，消息队列是有长度的，如果消息队列长度超过指定的长度，就直接抛弃；

2）秒杀的核心是处理任务，接收消息队列中的消息进行处理，这里的消息处理能力，取决于消费端本身的吞吐量；

**场景3：弱一致性事务模型中；**可以采用分布式消息队列的实现最大能力通知来实现最终一致性问题。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3p5MzQ1MjkzNzIx_size_16_color_FFFFFF_t_70 2][]

一致性示意图

### **ActiveMq的简介** ###

Activemq是完全基于jms规范实现的一个消息中间件产品，是Apache开源基金会研发的消息中间件。activemq主要应用在分布式系统结构中，帮助构建高可用，高性能，可伸缩的企业级面向消息服务的系统。

Activemq的特性

1）多语言和多协议

2）完全支持jms1.1和j2ee1.4规范

3） 对spring的支持，Activemq可以很容易内嵌到spring模块汇中

从jms规范来了解activemq

**      jms的定义**

java消息服务（java message service）是java平台开发关于面向消息中间件的api,用于两个应用程序之间，或者分布式系统中的发送消息，进行异步通信。

jms一个与具体平台无关的api,绝大多数mom（面向消息中间件）提供都对jms提供了支持。avtivemq就是其中的一种实现。

**     什么是mom?**

MOM是面向消息的中间件，使用消息传送提供者来协调消息传送操作，mom需要提供api和管理工作。客户端使用api调用，把消息发送到提供者管理的目的地，在发送消息之后，客户端会继续执行其他工作，并且在接收到这个消息确认之前，提供者一直保留该消息。

![20190225113435941.png][]

### **MOM的特点：** ###

1）消息异步接收，发送者不需要等待消息接受者响应；

2）消息可靠接收，确保消息在中间件可靠保存。只有接收方收到后才删除消息；

java消息发送服务规范最初的开发目的是为了使java应用程序能够访问现有MOM系统。引入该规范之后，她以被许多现有的MOM供应商凭借自己的需要实现异步消息传送系统。

### **JMS规范：** ###

jms规范的目的就是为了java应用程序能够访问现有MOM（消息中间件）系统，形成一套统一的标准规范，解决不同消息中间件之间的协作问题。在创建jms规范时，设计者系统能够结合现有的消息传送的精髓：

1）不同消息传播模式，如：点对点消息发送，发布/订阅消息传送；

2）提供接受同步和异步的消息工具；

3）对可靠消息传送的支持；

4）常见格式，如流，文本，字节；

### **jms的体系结构：** ###

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3p5MzQ1MjkzNzIx_size_16_color_FFFFFF_t_70 3][]

### **消息的传播流程：** ###

![20190225115100832.png][]

### ** jms的基本功能：** ###

消息域传递，jms的规范中，定义了两种消息传递域，点对点（point-to-point）消息传递域和发布/订阅（pub/sub）

简单理解，有点类似于通过qq聊天，在群里面发消息和给其中一个同学私聊消息。在群里发消息，所有的成员都能收到消息，私聊只能被私聊的学员能够收到消息。

点对点传递消息：

1）每个消息只能有一个消费者，

2）消息的生产者和消费者之间没有时间上的相关性，无论消费者在生产者发送消息时候任何状态，都可以提取消息。

![20190225120010769.png][]

发布订阅消息传递域：

1）每个消息可以有多个消费者

2）生产者和消费者之间有时间上的相关性。订阅一个主题的消费者只能消费自它订阅之后发布的消息。jms规范允许客户创建持久订阅，在一定程度上降低了时间上的相关性要求。持久订阅允许消费者它在未处于激活时发送的消息

![20190225153907245.png][]

### **消息的结构组成：** ###

jms消息的组成部分：消息头，属性，消息体

消息头（header）-消息头包含消息的识别信息和路由信息，消息头包含一些标准的属性，如：

jmsDestination       消息发送的目的地，queue或者topic

jmsDeliveryMode   传送模式，持久模式和非持久模式

jmsPriority             消息优先级（优先级分为10个级别，默认是4）

jmsMessageID       唯一识别每个消息的标识

属性

按类型可以分为应用设置的属性，标准属性和消息中间件定义的属性。

1）应用程序设置和添加的属性，比如Message.setStringProperty("key","value");通过下面的代码可以获得自定义属性的，在接收端的代码中编写。在发送端，定义消息属性：message.setStringProperty("zhangyu","Hello World");

在接收端接收数据

Enumeration  enumeration=message.getPropertyNames();
    while(enumeration.hasMoreElements()){
    String name=enumeration.nextElement().toString();
       System.out.println("name:"+name+":"+message.getStringProperty(name));
       System.out.println();
    }

2）jms定义的属性，使用jmsx作为属性的前缀，通过下面代码可以返回所有连接支持的jmsx属性的名字。

Enumeration names=connection.getMetaData().getJMSXPropertyNames();
    while(names.hasMoreElements()){
         String name=(String)names.nextElement();
         System.out.println(name);
    }

jms provider特定的属性:

消息体，就是我们要发送的内容，jms的api定义了五种消息格式，可以使用不同形式发送接收数据，并兼容所有的格式：

1））TextMessage    java.lang.String对象，如xml的内容

2））MapMessage    可以存储任何类型的数据

3）） BytesMessage  字节流

4）） StreamMessage  java中的输入输出流

5））ObjectMessage    java中可序列化的对象

6）） 没有消息体的对象

### **   持久订阅** ###

持久订阅的概念比较好理解，以qq为例，我们把qq退出时，下次登录的时候，仍然可以收到离线消息。

持久订阅就是这个道理，它有 两个特点：

1）持久订阅者和非持久订阅者针对的domin是pub/sub,而不是p2p；

2）当broker发送消息给订阅者时候，如果订阅者处于未激活状态：持续订阅者可以收到消息，而非持续订阅者就收不到消息；当然这种方式也有一定的影响：当持久订阅者处于未激活状态时，broker需要为持久订阅者保存消息，如果持久订阅者订阅的消息太多就会溢出。

消费端改动：

connection=connectionFactory.createConnnection();
    connection.setClientID("zhangyu_01");
    connection.start();
    Session session=connection.createSession(Boolean.TRUE,Session.AUTO_ACKNOWLEDGE);
    Topic destination=session.createTopic("mytopic");
    MessageComsumer=session.createDurableSubcriber(destination,"zhangyu_001");
    TextMasessage message=(TextMessage)consumer.receive();
    System.out.println(message.getText());

修改了三处，然后先启动消费端去注册一个持久化订阅，持久订阅时，客户端想Jms服务器注册一个自己身份的ID,当这个客户端处于离线时，jms provider会为这个ID保存所有发送到主题的消息，当客户端再次连接到Jms provider时，会根据自己的ID得到所有当自己处于离线时发送到主题的消息。

这个身份ID，在代码中体现是connection的ClientID,这个其实很好理解，你要想收到朋友发送的qq消息，前提是你得注册一个qq号码 ，还有一台能上网的电脑或手机，设备就相当于是clientid是唯一的；qq号相当于订阅者的名称，在同一台设备上，不能用同一个qq挂两个客户端。连接的clientid必须是唯一，订阅者的名称在同一连接必须唯一，这个才能唯一的确定连接和订阅者。

jsm消息的可靠机制

理论上说，我们需要保证消息中间件上的消息，只有被消费者确认以后才会被签收，相当于我们寄出了一个快递出去，收件人没有手打快递，就任务这个包裹还是属于待签收状态，这个才能保证包裹能够安全到收件人手里。消息中间件也是一样。

消息的消费通常包含3个阶段:客户接受消息，客户处理消息，消息被确认，首先来简单了解的事务性会话和非事务性会话的概念，jms session接口提供了commit和rollback方法，事务提交产生的所有消息被销毁，消费的所有消息被恢复并重新提交，除非他们已经过期。事务性的会话总是牵涉到事务处理中，commit或rollback方法一旦被调用，一个事务就结束了，而另一个事务被开始。关闭事务性会话将回滚其中的事务。

### **       在事务行会话中** ###

**     **  事务状态下进行发送操作，消息并未真正投递到中间件，而只是进行session.commit操作之后，消息才会发送到中间件，再转发到适当的消费者进行处理。如果是调用rollback操作，则表明，当前事务期间内所有发送的消息都取消掉。通过在创建session的时候，使用true or false来决定当前的会话是事务性还是非事务性。

connection.createSession(Boolean.tue,session.auto\_acknowledge);

在事务性会话中，消息的确认是自动进行，也就是通过session.commit()以后，消息会自动确认。

1）必须保证发送端和接收端都是事务性会话

**   在非事物型会话中**

** **  消息何时被确认取决于创建会话时的应答模式（ackonwledgement mode）,有三个可选项

Session.auto-ackonwledge当客户端成功的从receive方法返回的时候，或者从MessageLisenner.onMesage方法成功返回的时候，会话自动确认客户收到消息。

Session.client\_acknowledge客户端调用消息的acknowledge方法确认消息。发送端发送10个消息，接收端接收10个消息，接收到10个消息，但是在i=5的时候，调用message.acknowledge()进行确认，会发现0-4的消息都会被确认。

Session.dups\_acknowledge 消息延迟确认，指定消息提供者在消息接收者没有确认发送时重新发送消息，这个模式不在乎接收者收到重复的消息。

### **消息的持久化存储** ###

**     **   消息的持久化存储也是保证可靠性最重要的机制之一，也是消息发送到broker以后，如果broker出现故障宕机了，那么存储在broker上的消息不应该丢失。可以设置发送端的持久化和非持久化特性。

MessageProducer producer=Session.createProducer(destination);

producer.setDeliveryMode(DeliveryMode.PERISTMENT);

非持久化：jms provider不会将它存到文件/数据库等稳定的存储介质中。也就是说非持久化消息驻留在内存中 ，如果jms provider宕机，那么内存中的非持久化消息会丢失。

持久化：对于持久化消息，消息提供者会使用存储-转发机制，先将消息存储到稳定介质中，等消息发送成功后再删除，如jms provider 挂掉后，这些未送达的消息不会丢失，jms provider恢复正常后，会重新读取这些消息，并传递给消费者。

[github]: https://github.com/zhangyu345293721
[activemq_1--_activemq]: https://blog.csdn.net/zy345293721/article/details/87913000
[activemq_2---]: https://blog.csdn.net/zy345293721/article/details/87930437
[activemq_3--_2]: https://blog.csdn.net/zy345293721/article/details/87939302
[20190225105759824.png]: /images/20220317/49b2d8e59c5b46f68bd03fde905fe001.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3p5MzQ1MjkzNzIx_size_16_color_FFFFFF_t_70]: /images/20220317/4ea9c6bf6d8f4c3b807591097e2a3541.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3p5MzQ1MjkzNzIx_size_16_color_FFFFFF_t_70 1]: /images/20220317/f82f4cd7e61a49a48f20ce4a1fd8e98f.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3p5MzQ1MjkzNzIx_size_16_color_FFFFFF_t_70 2]: /images/20220317/bf426f01769c417b9553b337caf73627.png
[20190225113435941.png]: /images/20220317/40caa952cdab4fe39aca7b784422188c.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3p5MzQ1MjkzNzIx_size_16_color_FFFFFF_t_70 3]: /images/20220317/42d8127db4f049b3802a11a7320c50d6.png
[20190225115100832.png]: /images/20220317/612c4af686074312998cf3da28cba224.png
[20190225120010769.png]: /images/20220317/89e0769525424b7f8338bd43e1a1fdf8.png
[20190225153907245.png]: /images/20220317/4e70b85c6fb344239e028385fe2be8ec.png