1. JMS简介

1. JMS全称：Java Message Service ，中文：Java 消息服务，是 Java 的一套 API 标准
2. 最初的目的是为了使应用程序能够访问现有的 MOM系
3. MOM：Message Oriented Middleware，即消息中间件，它可以利用高效可靠的消息传递机 制进行平台无关的数据交流，并基于数据通信来进行分布式系统的集成
4. 常见 MOM 系统包括 Apache 的 ActiveMQ、 阿里巴巴的 RocketMQ、IBM 的 MQSeries、Microsoft 的 MSMQ、BEA 的 RabbitMQ 等。并非全部的 MOM 系统都遵循 JMS 规范，也就是并非全提供了JMS实现，提供了JMS 实现的 MOM，又被称为 JMSProvider
5. JMS与MOM的关系类似JDBC和数据库之间的关系

2 消息中间件的应用场景

1. 异步通信：消息队列提供了异步处理机制，允许用户把一个消息放入队列，但并不立即处理它
2. 过载保护：不将请求直接发送给服务端，而是由服务端自己来取，这样防止大量请求同时到达服务端，使整个系统崩溃
3. 解耦：A和B直接相连时，一旦B死掉，那么A的功能也都不好用了，采用消息中间件解耦，可以保证B死掉，A也能正常发送给队列，当B复活后，又可以继续完成之前队列中的任务
4. 消息通讯：客户端A，客户端B，客户端N订阅同一主题，进行消息发布和接收。实现类似聊天室效果
5. 顺序保证：在大多使用场景下，数据处理的顺序都很重要。大部分消息队列本来就是排序的，并且能保证数据会按照特定的顺序来处理。
6. 数据流处理：分布式系统产生的海量数据流，如：业务日志、监控数据、用户行为等，针对这些数据流进行实时或批量采集汇总，然后进行大数据分析是当前互联网的必备技术，通过消息队列完成此类数据收集是最好的选择
7. 扩展性：消息中间件可以很容易横向扩容

3 常用消息队列比较

特性MQ	ActiveMQ	RabbitMQ	RocketMQ	Kafka
生产者消费者模式	支持	支持	支持	支持
发布订阅模式	支持	支持	支持	支持
请求回应模式	支持	支持	不支持	不支持
Api完备性	高	高	高	高
多语言支持	支持	支持	java	支持
单机吞吐量	万级	万级	万级	十万级
消息延迟	无	微秒级	毫秒级	毫秒级
可用性	高（主从）	高（主从）	非常高（分布式）	非常高（分布式）
消息丢失	低	低	理论上不会丢失	理论上不会丢失
文档的完备性	高	高	高	高
提供快速入门	有	有	有	有
社区活跃度	高	高	有	高
商业支持	无	无	商业云	商业云

4 JMS中的一些角色

1. Broker：消息服务器，相当于server，提供消息核心服务
2. Provider：消息生产者，是由会话创建的一个对象，用于把消息发送到一个目的地
3. Consumer：消息消费者，是由会话创建的一个对象，它用于接收发送到目的地的消息
   1. 消费消息的两种方式：
      1. 同步消费：调用消费者的receive方法，从目的地中显式提取消息。receive方法可以一直阻塞到消息到达
      2. 异步消费：客户端可以为消费者注册一个消息监听器，以定义在消息到达时所采取的动作
4. p2p：基于点对点的消息模型
   1. 消息生产者生产消息发送到 queue 中，然后消息消费者从 queue 中取出并且消费消息
   2. 消息被消费以后，queue 中不再有存储，所以消息消费者不可能消费到已经被消费的消息
   3. Queue 支持存在多个消费者，但是对一个消息而言，只会有一个消费者可以消费、其它 的则不能消费此消息了
   4. 当消费者不存在时，消息会一直保存，直到有消费消费
      
5. pub/sub：基于订阅/发布的消息模型
   1. 消息生产者（发布）将消息发布到 topic 中，同时有多个消息消费者（订阅）消费该消
      息
   2. 和点对点方式不同，发布到 topic 的消息会被所有订阅者消费
   3. 当生产者发布消息，不管是否有消费者。都不会保存消息 一定要先有消息的消费者，后有消息的生产者
      
6. PTP 和 PUB/SUB 简单对比

1	Topic	Queue
	Publish Subscribe messaging 发布 订阅消息	Point-to-Point 点对点
有无状态	topic 数据默认不落地，是无状态的，也就是发消息时，如果接收的人不在线，那么该消息他就收不到了。	Queue 数据默认会在 mq 服 务器上以文件形式保存，比如 Active MQ 一 般 保 存 在 $AMQ_HOME\data\kahadb 下 面。也可以配置成 DB 存储。
完整性保障	并不保证 publisher 发布的每条数 据，Subscriber 都能接受到。	Queue 保证每条数据都能 被 receiver 接收。消息不超时。
消息是否会丢失	一般来说 publisher 发布消息到某 一个 topic 时，只有正在监听该 topic 地址的 sub 能够接收到消息；如果没 有 sub 在监听，该 topic 就丢失了。	Sender 发 送 消 息 到 目 标 Queue， receiver 可以异步接收这 个 Queue 上的消息。Queue 上的 消息如果暂时没有 receiver 来 取，也不会丢失。前提是消息不 超时。
消息发布接 收策略	一对多的消息发布接收策略，监 听同一个topic地址的多个sub都能收 到 publisher 发送的消息。Sub 接收完 通知 mq 服务器	一对一的消息发布接收策 略，一个 sender 发送的消息，只 能有一个 receiver 接收。 receiver 接收完后，通知 mq 服务器已接 收，mq 服务器对 queue 里的消 息采取删除或其他操作。

7. Queue：队列存储，常用与点对点消息模型 ，默认只能由唯一的一个消费者处理。一旦处理消息删除
8. Topic：主题存储，用于订阅/发布消息模型主题中的消息，会发送给所有的消费者同时处理。只有在消息可以重复处 理的业务场景中可使用。Queue/Topic都是 Destination 的子接口
9. ConnectionFactory ：连接工厂，jms中用它创建连接。连接工厂是客户用来创建连接的对象，例如ActiveMQ提供的ActiveMQConnectionFactory
10. Connection：JMS Connection封装了客户与JMS提供者之间的一个虚拟的连接
11. Destination：消息的目的地。在点对点消息传递域中，目的地被成为队列（queue）；在发布/订阅消息传递域中，目的地被成为主题（topic）
12. Session：JMS Session是生产和消费消息的一个单线程上下文。会话用于创建消息生产者（producer）、消息消费者（consumer）和消息（message）等。会话提供了一个事务性的上下文，在这个上下文中，一组发送和接收被组合到了一个原子操作中

5 JMS的消息格式

1. 消息头
2. 消息属性
3. 消息体

6 JMS的消息类型

1. TextMessage：文本消息
2. MapMessage：k-v
3. BytesMessage：字节流
4. StreamMessage：java原始的数据流
5. ObjectMessage：序列化的java对象

7 设置消息的几种特性

7.1 可靠性

1. 消息的成功消费通常包含三个阶段：客户接收消息、客户处理消息和消息被确认
2. 在事务性会话中，当一个事务被提交的时候，确认自动发生
3. 在非事务性会话中，消息何时被确认取决于创建会话时的应答模式（acknowledgement mode）。该参数有以下三个可选值：
   1. Session.AUTO_ACKNOWLEDGE：当客户成功的从receive方法返回的时候，或者从MessageListener.onMessage方法成功返回的时候，会话自动确认客户收到的消息
   2. Session.CLIENT_ACKNOWLEDGE：客户通过消息的acknowledge方法确认消息。需要注意的是，在这种模式中，确认是在会话层上进行：确认一个被消费的消息将自动确认所有已被会话消费的消息。例如，如果一个消息消费者消费了10个消息，然后确认第5个消息，那么所有10个消息都被确认
   3. Session.DUPS_ACKNOWLEDGE：该选择只是会话迟钝的确认消息的提交。如果JMS Provider失败，那么可能会导致一些重复的消息。如果是重复的消息，那么JMS Provider必须把消息头的JMSRedelivered字段设置为true。

7.2 持久性

1. JMS 支持以下两种消息提交模式
   1. PERSISTENT：指示JMS Provider持久保存消息，以保证消息不会因为JMS Provider的失败而丢失
   2. NON_PERSISTENT：不要求JMS Provider持久保存消息

7.2 KahaDB存储

3. KahaDB是默认的持久化策略，所有消息顺序添加到一个日志文件中，同时另外有一个索引文件记录指向这些日志的存储地址，还有一个事务日志用于消息回复操作。是一个专门针对消息持久化的解决方案,它对典型的消息使用模式进行了优化。
4. 在data/kahadb这个目录下，会生成四个文件，来完成消息持久化
   1. db.data 它是消息的索引文件，本质上是B-Tree（B树），使用B-Tree作为索引指向db-*.log里面存储的消息
   2. db.redo 用来进行消息恢复 *
   3. db-.log 存储消息内容。新的数据以APPEND的方式追加到日志文件末尾。属于顺序写入，因此消息存储是比较 快的。默认是32M，达到阀值会自动递增
   4. lock文件 锁，写入当前获得kahadb读写权限的broker ，用于在集群环境下的竞争处理
5. 配置

<persistenceAdapter> <!--directory:保存数据的目录;journalMaxFileLength:保存消息的文件大小，是每个数据文件大小，超过后滚动 --> <kahaDBdirectory="${activemq.data}/kahadb"journalMaxFileLength="16mb"/> </persistenceAdapter>

6. 特性
   1. 日志形式存储消息
   2. 消息索引以 B-Tree 结构存储，可以快速更新
   3. 完全支持 JMS 事务
   4. 支持多种恢复机制

7.5 JDBC存储

使用JDBC持久化方式，数据库默认会创建3个表，每个表的作用如下：
activemq_msgs：queue和topic的消息都存在这个表中
activemq_acks：存储持久订阅的信息和最后一个持久订阅接收的消息ID
activemq_lock：跟kahadb的lock文件类似，确保数据库在某一时刻只有一个broker在访问

ActiveMQ 将数据持久化到数据库中。

不指定具体的数据库。 可以使用任意的数据库 中。

本环节中使用 MySQL 数据库。 下述文件为 activemq.xml 配置文件部分内容。

首先定义一个 mysql-ds 的 MySQL 数据源，然后在 persistenceAdapter 节点中配置 jdbcPersistenceAdapter 并且引用刚才定义的数据源。

dataSource 指定持久化数据库的 bean，createTablesOnStartup 是否在启动的时候创建数 据表，默认值是 true，这样每次启动都会去创建数据表了，一般是第一次启动的时候设置为 true，之后改成 false。

Beans中添加

<bean id="mysql-ds" class="org.apache.commons.dbcp.BasicDataSource" destroy-method="close"> <property name="driverClassName" value="com.mysql.jdbc.Driver"/> 
<property name="url" value="jdbc:mysql://localhost/activemq?relaxAutoCommit=true"/> 
<property name="username" value="activemq"/>
<property name="password" value="activemq"/>
<property name="maxActive" value="200"/>
<property name="poolPreparedStatements" value="true"/> </bean>

修改persistenceAdapter

        <persistenceAdapter><!-- <kahaDB directory="${activemq.data}/kahadb"/> --><jdbcPersistenceAdapter dataSource="#mysql-ds" createTablesOnStartup="true" /> </persistenceAdapter>

依赖jar包

commons-dbcp commons-pool mysql-connector-java

4. 表字段解释
   activemq_acks：用于存储订阅关系。如果是持久化Topic，订阅者和服务器的订阅关系在这个表保存。
   主要的数据库字段如下：

container：消息的destination 
sub_dest：如果是使用static集群，这个字段会有集群其他系统的信息 
client_id：每个订阅者都必须有一个唯一的客户端id用以区分 
sub_name：订阅者名称 
selector：选择器，可以选择只消费满足条件的消息。条件可以用自定义属性实现，可支持多属性and和or操作 
last_acked_id：记录消费过的消息的id。

2：activemq_lock：在集群环境中才有用，只有一个Broker可以获得消息，称为Master Broker，其他的只能作为备份等待Master Broker不可用，才可能成为下一个Master Broker。这个表用于记录哪个Broker是当前的Master Broker。

3：activemq_msgs：用于存储消息，Queue和Topic都存储在这个表中。

id：自增的数据库主键 
container：消息的destination 
msgid_prod：消息发送者客户端的主键 
msg_seq：是发送消息的顺序，msgid_prod+msg_seq可以组成jms的messageid 
expiration：消息的过期时间，存储的是从1970-01-01到现在的毫秒数 
msg：消息本体的java序列化对象的二进制数据 
priority：优先级，从0-9，数值越大优先级越高 
xid:用于存储订阅关系。如果是持久化topic，订阅者和服务器的订阅关系在这个表保存。

7.3 优先级

1. 可以使用消息优先级来指示JMS Provider首先提交紧急的消息。优先级分10个级别，从0（最低）到9（最高）。如果不指定优先级，默认级别是4。需要注意的是，JMS Provider并不一定保证按照优先级的顺序提交消息。

7.4 本地事务

在一个JMS客户端，可以使用本地事务来组合消息的发送和接收。JMS Session接口提供了commit和rollback方法。事务提交意味着生产的所有消息被发送，消费的所有消息被确认；事务回滚意味着生产的所有消息被销毁，消费的所有消息被恢复并重新提交，除非它们已经过期。 事务性的会话总是牵涉到事务处理中，commit或rollback方法一旦被调用，一个事务就结束了，而另一个事务被开始。关闭事务性会话将回滚其中的事务。 需要注意的是，如果使用请求/回复机制，即发送一个消息，同时希望在同一个事务中等待接收该消息的回复，那么程序将被挂起，因为知道事务提交，发送操作才会真正执行。 需要注意的还有一个，消息的生产和消费不能包含在同一个事务中

8 HelloWorld

1. 下载ActiveMQ

http://activemq.apache.org/

2. 启动ActiveMQ

bin/win64/activemq.bat

3. 进入管理界面

http://localhost:8161/

4. 修改访问端口

conf/jetty.xml

5. 创建maven项目
   1. 会自动帮你下载、并引入需要的activemq-all-5.15.12.jar以及源码，如果创建正常项目，需要将E:\Program Files (x86)\apache-activemq-5.15.12\下的该jar包引入
   2. File–new–Project–Maven Project–选中create a simple project–填写Goup Id与Artifact ID
   3. 修改pom.xml文件

<project xmlns="http://maven.apache.org/POM/4.0.0"xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd"><modelVersion>4.0.0</modelVersion><groupId>com.mashibing.mq</groupId><artifactId>activemq02</artifactId><version>0.0.1-SNAPSHOT</version><!-- https://mvnrepository.com/artifact/org.apache.activemq/activemq-all --><dependencies><dependency><groupId>org.apache.activemq</groupId><artifactId>activemq-all</artifactId><version>5.15.12</version></dependency></dependencies>
</project>

6. Sender

package com.mashibing.mq;
import javax.jms.Connection;
import javax.jms.DeliveryMode;
import javax.jms.MessageProducer;
import javax.jms.Queue;
import javax.jms.Session;
import javax.jms.TextMessage;import org.apache.activemq.ActiveMQConnectionFactory;/*** 消息发送* @author 一明哥**/
public class Sender {public static void main(String[] args) throws Exception{// 1.获取连接工厂ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory("admin","admin","tcp://localhost:61616");// 2.获取一个向ActiveMQ的连接Connection connection = connectionFactory.createConnection();// 3.获取sessionSession session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);// 4. 找目的地，获取destination，消费端，也会从这个目的地取消息Queue queue = session.createQueue("user");// 51.消息创建者MessageProducer producer = session.createProducer(queue);//	producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);// consumer -> 消费者// producer -> 创建者// 5.2. 创建消息for (int i = 0; i < 1000; i++) {TextMessage textMessage = session.createTextMessage("hi: " + i);// 5.3 向目的地写入消息if(i % 4 == 0) {// 设置消息的优先级// 对producer 整体设置//	producer.setPriority(9);//	producer.send(textMessage,DeliveryMode.PERSISTENT,9,1000 * 100);textMessage.setJMSPriority(9);}		producer.send(textMessage);Thread.sleep(3000);}	// 6.关闭连接connection.close();	System.out.println("System exit....");}
}

7. Receiver

package com.mashibing.activemq01;import javax.jms.Connection;
import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageConsumer;
import javax.jms.MessageProducer;
import javax.jms.Queue;
import javax.jms.Session;
import javax.jms.TextMessage;import org.apache.activemq.ActiveMQConnectionFactory;public class Receiver {public static void main(String[] args) throws Exception {// 1. 建立工厂对象，ActiveMQConnectionFactory activeMQConnectionFactory = new ActiveMQConnectionFactory(ActiveMQConnectionFactory.DEFAULT_USER,ActiveMQConnectionFactory.DEFAULT_PASSWORD,"tcp://localhost:61616");//2 从工厂里拿一个连接Connection connection = activeMQConnectionFactory.createConnection();connection.start();//3 从连接中获取Session(会话)Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);// 从会话中获取目的地(Destination)消费者会从这个目的地取消息Queue queue = session.createQueue("f");//从会话中创建消息提供者MessageConsumer consumer = session.createConsumer(queue);//从会话中创建文本消息(也可以创建其它类型的消息体)while (true) {TextMessage receive = (TextMessage)consumer.receive();System.out.println("TextMessage:" + receive.getText());}}
}

9 Active MQ的安全机制

1. web控制台安全：重启后生效

//conf/jetty.xml中id="securityLoginService">配置了其登录相关用户的配置文件
conf/jetty-realm.properties

//用户名：密码，角色
admin: admin, admin
user: user, user

2. 消息安全机制：配置建立连接工厂时的用户和密码

conf/activemq.xml

<plugins><simpleAuthenticationPlugin><users><authenticationUser username="admin" password="admin" groups="admins,publishers,consumers"/><authenticationUser username="publisher" password="publisher"  groups="publishers,consumers"/><authenticationUser username="consumer" password="consumer" groups="consumers"/><authenticationUser username="guest" password="guest"  groups="guests"/></users></simpleAuthenticationPlugin></plugins>