ActiveMQ 消息中间件

浅浅的花香味﹌ 2022-03-18 01:40 326阅读 0赞

### ActiveMQ简单介绍与使用 ###

*  1 JMS
 *   *  1.1 背景&JMS概述
     *  1.2JMS术语
     *  1.3JMS术语概念（一）
     *  1.6 消息格式定义
 *  2 ActiveMQ
 *   *  2.1 ActiveMQ简介
     *  2.2 ActiveMQ使用
     *  2.3 ActiveMQ Hello World
     *  2.4 ActiveMQ安全机制
 *  3 API对象
 *   *  3.1 Connection对象的使用
     *  3.2 Session对象的使用
     *  3.3 MessageProducer对象的使用
     *   *  3.3.2 ActiveMQ独占消费者（Exclusive Consumer）
     *  3.4 MessageConsumer（一）
     *  3.4 MessageConsumer（二）
     *  3.5 Message
     *  3.6 创建临时消息
 *  4. 主题
 *   *  4.1 高级主题（p2p）
     *  4.2 高级主题（publish-subscribe 发布订阅）
     *  4.3 高级主题（与spring进行整合）
     *  4.4 高级主题（ActiveMQ集群）

# 1 JMS #

## 1.1 背景&JMS概述 ##

当前，CORBA、DCOM、RMI等RPC中间件技术已经广泛应用于各个领域。但是面对规模和复杂度都越来越高的分布式系统，这些技术也是显示出`局限性`：  
（1）`同步通信`： 客户发出调用后，必须等待服务对象完成处理并返回结果后才能继续执行；  
（2）`客户和服务对象的生命周期紧密耦合`： 客户进程和服务对象进程都必须正常运行；如果由于服务对象崩溃或者网络故障导致客户的请求不可达，客户会接受到异常；

（3）`点对点通信`： 客户的一次调用只发送给某个单独的目标对象

面向消息的中间件（Message Oriented Middleware，MOM）较好的解决了以上问题。发送者将消息发送给消息服务器，消息服务器将消息存放在若干队列中，在合适的时候再将消息转发给接受者。  
这种模式下，发送和接受都是异步的，发送者无需等待；  
两者的生命周期未必相同：发送消息的时候接收者不一定运行，接受消息的时候发送者也不一定运行；  
一对多通讯：对于一个消息可以有多个接收者

JAVA消息服务（JMS）定义来了Java中访问消息中间件的接口。JMS只是接口，并没有给与实现，实现JMS接口的消息中间件称为JMS Provider，已有的MOM系统包括Apache的ActiveMQ，以及阿里巴巴的RocketMQ、IBM的MQSeries、Microsoft的MSMQ和BEA的MessageQ、RabbitMQ、Kafka等等…他们基本都遵循JMS规范

**应用场景**  
[https://blog.csdn.net/cws1214/article/details/52922267][https_blog.csdn.net_cws1214_article_details_52922267]

## 1.2JMS术语 ##

JMS实现JMS接口的消息中间件  
Provider（MessageProvider）: 生产者  
Consumer（MessageConsumer）: 消费者  
PTP： Point to Point，即点对点的消息模型；  
Pub/Sub: Publish/Subscribe,即发布/订阅的消息模型;  
Queue: 队列目标；  
Topic： 主题目标；  
ConnectionFactory：连接工厂，JMS用它创建链接；  
Connection： JMS客户端到JMS Provider的连接；  
Destination： 消息的目的地；  
Session： 会话，一个发送或接受消息的线程；

## 1.3JMS术语概念（一） ##

***ConnectionFactory接口（链接工厂）***  
用户用来创建到JMS提供者的链接的被管对象。JMS客户通过可移植的接口访问连接，这样当下层的实现改变时，代码不需要进行修改。管理员在JNDI名字空间中配置链接工厂，这样，JMS客户才能够查找到它们。根据消息类型的不同，用户将使用队列链接工厂，或者主题连接工厂。

***Connection接口（连接）***  
连接代表了应用程序和消息服务器之间的通信链路。在获得了连接工厂后，就可以创建一个与JMS提供者的连接。根据不同的连接类型，连接允许用户创建会话，可以送和接受队列和主题到目标

***Destination接口（目标）***  
目标是一个包装了消息目标标识符的被管对象，消息目标是指消息发布和接受的地点，或者是队列，或者是主题。JMS管理员创建这些对象，然后用户通过JNDI发现它们。和链接工厂一样，管理员可以创建两种类型的目标，点对点模型的队列，以及发布者/订阅者模型的主题。

***Session接口（会话）***  
表示一个单线程的上下文，用于发送和接受消息。由于会话是单线程的，所以消息是连续的，就是说消息是按照发送的顺序一个一个接受的。会话的好处是他支持事物。如果用户选择了事物支持，会话上下文将保存一组消息，直到事物被提交才发送这些消息。在提交事物之前，用户可以使用回滚操作取消这些消息。一个会话允许用户创建消息生产者来发送消息，创建消息消费者来接受消息。

## 1.6 消息格式定义 ##

JMS定义了吴总不同的消息正文格式，以及调用的消息类型，允许你发送并接受以一些不同形式的数据，提供现有消息格式的一些级别的兼容性。  
  
• StreamMessage Java原始值的数据流  
• MapMessage 一套名称–值对  
• TextMessae 一个字符串对象  
• ObjectMessage 一个序列化的Java对象  
• BytesMessage 一个未解释字节的数据流

# 2 ActiveMQ #

## 2.1 ActiveMQ简介 ##

ActiveMQ是Apache出品，最流行的，能力强劲的开源的消息总线。  
ActiveMQ是一个完全支持JMS1.1和J2EE1.4规范的JMS Provider实现，尽管JMS规范出台已经是很久的事情了，但是JMS在当今的J2EE应用中间任然扮演者特殊的地位，可以说ActiveMQ在业界应用最广泛，当然如果想要有更强大的性能和海量数据处理能力，ActiveMQ还需要不断地升级版本，80%以上的业务我们使用ActiveMQ足够满足需求，当然后续天猫、淘宝网这种大型的电商网站，尤其是双11这种特殊事件，ActiveMQ需要进行很复杂的优化源码以及架构设置才能完成，我们之后会学习一个更强大的分布式消息中间件，RocketMQ，可以说ActiveMQ是核心，是基础，所以我们必须要掌握好。

**支持：队列数量一两万，消息数一天几十万条**

## 2.2 ActiveMQ使用 ##

去官方网站下载： [http://activemq.apache.org/activemq-5153-release.html][http_activemq.apache.org_activemq-5153-release.html]  
目前最新版本是：apache-activemq-5.15.3-bin.zip  
我们学习暂时使用的是Windows 系统，后期学习RocketMQ的时候使用Linux系统。  
下载好进行解压，目录如下：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FpYXppbGlwaW5n_size_16_color_FFFFFF_t_70]

## 2.3 ActiveMQ Hello World ##

如：activeMQ-API项目  
我们首先写一个简单的Hello World示例，让大家感受ActiveMQ，我们需要实现接受者和发送者两部分代码的编写。  
Sender/Receiver

1.  第一步：  
    建立ConnectionFactory工厂对象，需要填入用户名、密码、以及要连接的地址，均使用默认即可，默认端口为”tcp:localhost:61616”
2.  第二步：  
    通过ConnectionFactory工厂对象创建一个Connection连接，并且调用Connection的start方法开启连接，Connection默认关闭的。
3.  第三步：  
    同步Connection对象创建Session会话（上下文环境对象），用于接受消息，参数配置1为是否启用事物，参数配置2为签收模式，一般我们设置自动签收。
4.  第四步：  
    通过Session创建Destination对象，指的是一个客户端用来指定生产消息目标和消费消息来源的对象，在PTP模式中，Destination被称作Queue即队列；在Pub/Sub模式，Destination被称作Topic即主题。在程序中可以使用多个Queue和Topic。
5.  第五步：我们需要通过Session对象创建消息的发送和接受对象（生产者和消费者）MessageProducer/MessageConsumer
6.  第六步：我们可以使用MessageProducer的setDeliveryMode方法为其设置持久化和非持久化特性（DeliveryMode），我们销售消息介绍。
7.  第七步：最后我们使用JMS规范的TextMessage形式创建数据（通过Session对象），并用MessageProducer的send方法发送数据。同理客户端使用receive方法进行接收数据，最后不要忘记关闭Connection连接。

## 2.4 ActiveMQ安全机制 ##

→ ActiveMQ的web管理界面： `http:localhost:8161/admin`  
→ ActiveMQ管控台使用jetty部署，所以需要修改密码则需要到相应的配置文件  
→ D:\\lp\_data\\activemq-notes\\apache-activemq-5.15.3\\conf\\jetty-realm.properties

→ ActiveMQ应该设置有安全机制，只有符合认证的用户才能进行发送和获取消息，所以我们也可以在activemq.xml里去添加安全验证配置！  
→ D:\\lp\_data\\activemq-notes\\apache-activemq-5.15.3\\conf\\activemq.xml,在第123行之后添加配置！（添加一个插件配置即可）

<plugins>
    	<simpleAuthenticationPlugin>
            <users>
                <authenticationUser username="liping" password="liping" groups="admins,publishers,consumers"/>
            </users>
        </simpleAuthenticationPlugin>
    </plugins>

修改java代码如下

ConnectionFactory factory = new ActiveMQConnectionFactory(
    				"liping", //用户名
    				"liping", //密码
    				//ActiveMQConnectionFactory.DEFAULT_USER, 
    				//ActiveMQConnectionFactory.DEFAULT_PASSWORD, 
    				"tcp://localhost:61616");

→ ActiveMQ持久化存储： 可以切换不同的存储技术（默认是kadb、leveldb、mysql、oracle）  
→ 引入三个包：mysql驱动、dbcp和commons-pool三个jar  
**…conf/activemq.xml配置如下**

<bean id="mysql_conf" class="org.apache.commons.dbcp.BasicDataSource">
    	<property name="driverClassName" value="com.mysql.jdbc.Driver" />
    	<property name="url" value="jdbc:mysql://127.0.0.1:3306/test?relaxAutoCommit=true" />
    	<property name="username" value="root" />
    	<property name="password" value="123456" />
    	<property name="poolPreparedStatements" value="true"/>  
    </bean>
    
    
    <persistenceAdapter>
        
    	
    	<jdbcPersistenceAdapter dataSource="#mysql_conf"/>  
    </persistenceAdapter>

***修改为mysql持久化需要的jar包，将jar放在active服务器的lib下面：***  
mysql-connector-java-5.1.36.jar  
commons-dbcp-1.4.jar  
commons-pool-1.6.jar  
如下图：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FpYXppbGlwaW5n_size_16_color_FFFFFF_t_70 1]

# 3 API对象 #

## 3.1 Connection对象的使用 ##

在成功创建正确的ConnectionFactory后，下一步将是创建一个链接，他是JMS定义的一个接口。ConnectionFactory负责返回可以与底层消息传递系统进行通信的Connection实现。通常客户端只使用单一连接。根据JMS文档，Connection的目的是“利用JMS提供者封装开发的连接”，以及表示“客户端与提供者服务例程之间的开放TCP/IP套接字”。改文档还支出Connection应该是进行客户端身份验证的地方等等。  
当一个Connection被创建是，他的传输默认是关闭的，必须使用start方法开启。一个Connection可以创建一个或多个的Session。  
当一个程序执行完成后，必须关闭之前创建的Connection，否则ActiveMQ不能释放资源，关闭一个Connection同样也关闭了Session，MessageProducer和MessageConsumer。  
Connection createConnection();  
Connection createConnection(String userName,String password);

## 3.2 Session对象的使用 ##

一旦从ConnectionFactory中获取一个Connection，必须从Connection中创建一个或者多个Session。Session是一个发送或接受消息的线程，可以使用Session创建MessageProducer，MessageConsumer和Message。  
Session可以被事物化，也可以不被事物话，通常，可以通过向Connection上的适当创建方法传递一个布尔参数对此进行设置。  
Session createSession(boolean transacted,int acknowledgeModel)  
其中transacted为使用事务标识，acknowledgeModel为签收模式。  
借宿事务的两种方法：提交或者回滚。当一个事物提交，消息被处理。如果事务中有一个步骤失败，事务就回滚，这个事务中的已执行的动作将被。在发送消息最后也必须要使用session.commit()方法表示提交事务。

三种签收模式：  
Session.AUTO\_ACKNOWLEDGE \# 当客户端从receive或onMessage成功返回时，Session自动签收客户端的这表消息的收条  
Session.CLIENT\_ACKNOWLEDGE \# 客户端调用Message的acknowledge()方法进行签收。在这种情况下，签收发生在Session层面；签收一个已消费的消息会自动地签收这个Session所有的已消费的收条。  
Session.DUPS\_OK\_ACKNOWLEDGE \# 此选项只是Session不必确保对传送消息的签收。他可以引起消息的重复，但是降低了Session的开销，多以只有客户端能容忍重复的消息，才可以使用

## 3.3 MessageProducer对象的使用 ##

MessageProducer： 是由Session创建的对象，用来想Destination发送消息。  
MessageProducer重载了四个send（）方法：

1、void send(Destination destination,Message msg);

2、void send(Destination destination,Message msg,int deliveryModel,int priority,long timeToLive);

3、void send(Message msg);

4、void send(Message msg,int deliveryModel,int priority,long timeToLive);

*  参数1 Destination：目标地址
 *  参数2 Message ：具体的数据信息
 *  参数3 deliveryModel：传送的数据模式（PERSISTENT持久和NON\_PERSISTENT非持久）如果容忍数据丢失，那么使用非持久，消息可以改善性能和减少存储的开销
 *  默认情况下使用的是持久传输。
 *  **持久传输和非持久传输最大的区别是：** 采用持久传输时，传输的消息会保存到磁盘中(messages are persisted to disk/database)，即“存储转发”方式。先把消息存储到磁盘中，然后再将消息“转发”给订阅者。  
    采用非持久传输时，发送的消息不会存储到磁盘中。  
    采用持久传输时，当Borker宕机 恢复后，消息还在。采用非持久传输，Borker宕机重启后，消息丢失。比如，当生产者将消息投递给Broker后，Broker将该消息存储到磁盘中，在Broker将消息发送给Subscriber之前，Broker宕机了，如果采用持久传输，Broker重启后，从磁盘中读出消息再传递给Subscriber；如果采用非持久传输，这条消息就丢失了。
 *  参数4 priority：优先级 \# 消息的优先级从0-9是个级别，0-4为普通消息，5-9是加急消息。如果不指定优先级，则默认为4。JMS不要求严格按照这是个优先级发送消息，但必须保证单次加急消息要先于普通消息达到，但也仅仅是先到达，不能保证顺序消费机制。默认情况下，消息永不会过期。  
    优先级需要在…conf/activemq.xml的`<broker >`标签配置如下

*  参数5 timeToLive：消息过期时间 \#如果消息在特定周期内失去意义，那么可以设置过期时间，时间单位为毫秒

### 3.3.2 ActiveMQ独占消费者（Exclusive Consumer） ###

我们经常希望维持队列中的消息，按一定次序转发给消息者。然而当有多个JMS Session和消息消费者实例的从同一个队列中获取消息的时候，就不能保证消息顺序处理。因为消息被多个不同线程并发处理着。  
在ActiveMQ4.x中可以采用Exclusive Consumer或者Exclusive Queues，避免这种情况，Broker会从消息队列中，一次发送消息给一个消息消费者来保证顺序。  
配置如下(三种配置方式)：

*  在brokerUrl中指定，默认值为false，对connectionFactory下所有的Queue有效  
    tcp://localhost:61616?jms.exclusiveConsumer=true  
    例：ConnectionFactory factory = new ActiveMQConnectionFactory(  
    “liping”, //用户名  
    “liping”,  
    “tcp://localhost:61616?jms.exclusiveConsumer=true”);
 *  通过api的方式，为特定的Queue下的消费者设定此值，在destinationUri中设定，只对当前Queue有效。  
    queueName?consumer.exclusive=true  
    例： queue = new ActiveMQQueue(“QUEUE\_NAME?consumer.exclusive=true”); consumer =  
    session.createConsumer(queue);
 *  broker端还支持一个参数“allConsumersExclusiveByDefault”，如果此参数为true，那么当前Queue中的所有消费者默认为exclusive  
    例：需要在…conf/activemq.xml的`<broker >`标签配置如下

*  A．  
    当在接收信息的时候有一个或者多个备份接收消息者和一个独占消息者的同时接收时候，无论两者创建先后，在接收的时候，均为独占消息者接收。
 *  B．  
    当在接收信息的时候，有多个独占消费者的时候，只有一个独占消费者可以接收到消息。
 *  C．  
    当有多个备份消息者和多个独占消费者的时候，当所有的独占消费者均close的时候，只有一个备份消费者接到到消息。  
    备注：备份消费者为不带任何参数的消费者。

## 3.4 MessageConsumer（一） ##

MessageConsumer是由Session创建的对象，用来从Destination接收消息。

Session中创建consumer的方法如下：  
MessageConsumer createConsumer(Destination destination);  
MessageConsumer createConsumer(Destination destination, String messageSelector); //消息选择器  
MessageConsumer createConsumer(Destination destination, String messageSelector, boolean noLocal) ;

TopicSubscriber createDurableSubscriber(Topic topic, String name);

TopicSubscriber createDurableSubscriber(Topic topic, String name, String messageSelector, boolean noLocal);

其中messageSelector为消息选择器；  
noLocal标志默认为false，当设置为true时限制消费者只能接收和自己相同的连接（Connection）所发布的消息，此标志只适用于主题，不适用于队列；  
name标识订阅主题所对应的订阅名称，持久订阅是需要设置此参数；

Public final String SELECTOR = “JMS\_TYPE = ‘MY\_TAG1’ ”;该选择器检查了传入消息的JMS\_TYPE属性，并确定了这个属性的值是否等于MY\_TAG1。如果相等，则消息被消费；如果不相等，那么消息会被忽略。

## 3.4 MessageConsumer（二） ##

消息的同步和异步接收：  
消息的同步接收是指客户端主动去接收消息，客户端可以采用MessageConsumer的receive方法接受下一个消息。  
Message receive();  
Message receive(long timeout);  
Message receiveNoWait();

消息的异步接收是指当消息到达时，ActiveMQ主动通知客户端，可以通过注册一个实现MessageListener接口的对象到MessageConsumer。MessageListener只有一个必须实现的方法— onMessage，他只接受一个参数，即Message。在为每个发送到Destination的消息实现onMessage时，将调用该方法。

## 3.5 Message ##

JMS程序的最终目的是生产和消费的消息能被其他程序使用，JMS的Message是一个既简单又不乏灵活的基本格式，允许创建不同平台上符合非JMS程序格式的消息。Message由一下几部分组成： 消息头，属性和消息体。

public abstract BytesMessage createBytesMessage()
            throws JMSException;
            
    public abstract MapMessage createMapMessage()
        throws JMSException;
    
    public abstract Message createMessage()
        throws JMSException;
    
    public abstract ObjectMessage createObjectMessage()
        throws JMSException;
    
    public abstract ObjectMessage createObjectMessage(Serializable Object)
        throws JMSException;
    
    public abstract StreamMessage createStreamMessage()
        throws JMSException;
    
    public abstract TextMessage createTextMessage()
        throws JMSException;
    
    public abstract TextMessage createTextMessage(String text)
        throws JMSException;

我们一般会在接收端通过instanceof方法区别数据类型

## 3.6 创建临时消息 ##

ActiveMQ通过createtemporaryQueue和createTemporaryTopic创建临时目标，这些目标持续到创建它的Connection关闭。只有创建临时目标的Connection所创建的客户端才可以从临时目标中接收消息，但是任何的生产者都可以向临时目标中发送消息。如果关闭了创建此目标的Connection，那么临时目标被关闭，内容也将消失。

TemporaryQueue createTemporaryQueue();  
TemporaryTopic createTemporaryTopic();

# 4. 主题 #

## 4.1 高级主题（p2p） ##

p2p的过程理解起来更加简单。他好比是两个人打电话，这两个人是独享这一条通信链路的。一方发送消息，另外一方接收，就这么简单。在实际应用中因为有多个用户对使用p2p链路，他的通信场景如下所示：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FpYXppbGlwaW5n_size_16_color_FFFFFF_t_70 2]

## 4.2 高级主题（publish-subscribe 发布订阅） ##

发布订阅模式有点类似于我们日常生活中订阅报纸。每年到年尾的时候，邮局就会发一本报纸集合让我们来选择订阅哪一个。在这个表里头列了所有出版发行的报纸，那么对于我们每一个订阅者来说，我们可以选择一份或者多份报纸。比如北京日报、潇湘晨报等。那么这些个我们订阅的报纸，就相当于发布订阅模式里的topic。有很多个人订阅报纸，也有人可能和我订阅了相同的报纸。那么，在这里，相当于我们在同一个topic里注册了。对于一份报纸发行方来说，它和所有的订阅者就构成了一个1对多的关系。这种关系如下图所示：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FpYXppbGlwaW5n_size_16_color_FFFFFF_t_70 3]

## 4.3 高级主题（与spring进行整合） ##

–> ActiveMQ可以很轻松的与Spring进行整合，Spring提供了一系列的接口类，非常的好用！

–> 我们接下来使用spring框架整合ActiveMQ，理由消息中间件，异步处理任务的机制，比如异步消费数据、异步发送邮件、异步做查询操作等…

–> [http://blog.csdn.net/jiangxuchen/article/details/8004570][http_blog.csdn.net_jiangxuchen_article_details_8004570] ActiveMQ集群

## 4.4 高级主题（ActiveMQ集群） ##

–> Zookeeper + LevelDB +ActiveMQ搭建集群，提供高可用操作

–> 官方文档： [http://activemq.apache.org/replicated-leveldb-store.html][http_activemq.apache.org_replicated-leveldb-store.html]

–> 步骤：详细见ActiveMQ集群使用文档

[https_blog.csdn.net_cws1214_article_details_52922267]: https://blog.csdn.net/cws1214/article/details/52922267
[http_activemq.apache.org_activemq-5153-release.html]: http://activemq.apache.org/activemq-5153-release.html
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FpYXppbGlwaW5n_size_16_color_FFFFFF_t_70]: /images/20220318/0897aa2bcf9f4401bb91521652371508.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FpYXppbGlwaW5n_size_16_color_FFFFFF_t_70 1]: /images/20220318/66f890cb590d4fe790c18cb6d1ab59ca.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FpYXppbGlwaW5n_size_16_color_FFFFFF_t_70 2]: /images/20220318/ff5b09a5c9804630b4beee00859ce235.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FpYXppbGlwaW5n_size_16_color_FFFFFF_t_70 3]: /images/20220318/788f4423b3004543af7585a93611d225.png
[http_blog.csdn.net_jiangxuchen_article_details_8004570]: http://blog.csdn.net/jiangxuchen/article/details/8004570
[http_activemq.apache.org_replicated-leveldb-store.html]: http://activemq.apache.org/replicated-leveldb-store.html