Java消息中间件之ActiveMQ
一、消息中间件
1.消息中间件概述
中间件：非底层操作系统软件，非业务应用软件，不是直接给最终用户使用的，不能直接给客户带来价值的软件统称为中间件。 消息中间件：关注与数据的发送和接受，利用高效可靠的异步消息传递机制集成分布式系统。 JMS：Java消息服务（Java Message Service）即JMS，是一个Java平台中关于面向消息中间件的API，用于在两个应用程序之间，或分布式系统中发送消息，进行异步通信。 AMQP：AMQP（advanced message queuing protocol）是一个提供统一消息服务的应用层标准协议，基于此协议的客户端与消息中间件可传递消息，并不受客户端/中间件不同产品，不同开发语言等条件的限制。 JMS和AMQP对比2.消息中间件图示
3.常见消息中间件对比
3.1.ActiveMQ
（1）概述：ActiveMQ是Apache出品，最流行的，能力强劲的开源消息总线。ActiveMQ是一个完全支持JMS 1.1和J2EE 1.4规范的，JMS Provider实现，尽管JMS规范出台已经很久的事情了，但是JMS在当今的J2EE应用中间仍然扮演着特殊的地位。
（2）特性：
多种语言和协议编写客户端。语言：Java、C、C++、C#、Ruby、Perl、Python、PHP。应用协议：OpenWire、Stomp REST、WS、Notification、XMPP、AMQP 完全支持JMS 1.1和J2EE规范（持久化，XA消息，事务） 虚拟主题，组合目的，镜像队列3.2.RabbitMQ
（1）概述：RabbitMQ是一个开源的AMQP实现，服务器端用Erlang语言编写。用于分布式系统中存储转发消息，在易用性、扩展性、高可用性等方面表现不俗。
（2）特性：
支持多种客户端，如：Python、Ruby、.NET、Java、JMS、C、PHP、ActionScript等 AMQP的完整实现（vhost虚拟主机、Exchange交换器、Binding绑定、Routing Key路由器等） 事务支持/发布确认 消息持久化3.3.Kafka
（1）概述：Kafka是一种高吞吐量的分布式发布订阅消息系统，是一个分布式的、分区的、可靠的分布式日志存储服务。它通过一种独一无二的设计提供了一个消息系统的功能。
（2）特性：
通过O（1）的算法复杂度的磁盘数据结构提供消息的持久化，这种结构对于即使数以TB的消息存储也能够保持长时间的稳定性能 高吞吐量：即使是非常普通的硬件Kafka也可以支持每秒钟数百万的消息 Partition、Consumer Group3.4.综合评价
二、JMS
1.JMS规范
（1）JMS相关概念
提供者：实现JMS规范的消息中间件服务器 客户端：发送或接收消息的应用程序 生产者/发布者：创建并发送消息的客户端 消费者/订阅者：接收并处理消息的客户端 消息：应用程序之间传递的数据内容 消息模式：在客户端之间传递消息的方式，JMS中定义了主题和队列两种模式（2）JMS消息模式
A.队列模型
客户端包括生产者和消费者 队列中的消息只能被一个消费者消费 消费者可以随时消费队列中的消息队列模型消息示意图：
B.主题模型
客户端包括发布者和订阅者 主题中的消息被所有订阅者消费 消费者不能消费订阅之前就发送到主题中的消息，即只能先订阅才能消费主题模型消息示意图：
（3）JMS编码接口
ConnectionFactory 用于创建连接到消息中间件的链接工厂 Connection 代表了应用程序和消息服务器之间的通信链路 Destination 指消息发布和接收的地点，包括队列或主题 Session 表示一个单线程的上下文，用于发送和接收消息 MessageConsumer 由会话创建，用于接收发送到目标的消息 MessageProducer 由会话创建，用于发送消息到目标 Message 是在消费者和生产者之间传送的对象，消息头，一组消息属性，一个消息体5.JMS编程接口之间的关系
2.ActiveMQ的安装与启动
（1）在Windows平台安装ActiveMQ
下载安装包 http://activemq.apache.org/ 直接启动，以管理员身份运行启动你电脑对应版本的activemq.bat，即可成功启动，比如，我的是64位的，即选择如下图中的win64文件夹下的activemq.bat，或者安装成Windows服务，以服务方式启动，记住使用以管理员身份运行。使用服务启动，我习惯设置为手动启动，你也可以设置为自动，即在开机时就启动
无论你是如何启动，启动成功后，使用浏览器访问127.0.0.1:8161就可以访问ActiveMQ的管理页面，点击Manage ActiveMQ broker，出现弹框输入用户名密码，都默认为admin，然后就可以进入管理主页了！
（2）在Linux平台安装ActiveMQ
下载安装包 wget http://mirror.bit.edu.cn/apache//activemq/5.15.2/apache-activemq-5.15.2-bin.tar.gz 解压安装包 tar -zxvf apache-activemq-5.15.2-bin.tar.gz 启动 解压完成后，进入解压后产生的目录，使用 ./activemq start 启动，可以访问linux服务器的ip地址加端口8161，能进入管理页面即成功安装并启动，使用./activemq stop 关闭。3.JMS的代码实现
3.1.使用JMS接口规范连接ActiveMQ
创建生产者 创建发布者 创建消费者 创建订阅者具体代码实现可参考本人码云项目：https://gitee.com/kevinshaw/jms-demo.git
3.2.使用Spring集成JMS连接ActiveMQ
（1）ConnectionFactory 用于管理连接的连接工厂
一个Spring为我们提供的连接池 JmsTemplate每次发消息都会重新创建连接，会话和productor Spring中提供SingleConnectionFactory和CachingConnectionFactory，SingleConnectionFactory对于建立连接请求只会返回同一个Connection，并且使用同一个close方法；CachingConnectionFactory继承自SingleConnectionFactory，所有拥有SingleConnectionFactory的所有功能，而且新增缓存功能，可以缓存会话，Producer，Consumer（2）JmsTemplate 用于发送和接收消息的模板类
是Spring提供的，只需要向Spring容器内注册这个类就可以使用JmsTemplate方便的操作jms JmsTemplate类是线程安全的，可以在整个应用范围使用（3）MessageListerner 消息监听器
实现一个onMessage方法，该方法只接收一个Message参数具体代码实现可参考本人码云项目：https://gitee.com/kevinshaw/jms-demo.git
三、ActiveMQ集群配置
1.对消息中间件群集的原因：
实现高可用，以排除单点故障引起的服务中断 实现负载均衡，以提升效率为更多客户提供服务2.ActiveMQ集群基础知识
（1）集群方式
客户端集群：让多个消费者消费同一队列，在队列模式下已经支持这种情况了，但是在主题模式下，多个消费者是消费了完整的消息，造成消息重复的可能； Broker clusters：多个Broker之间同步消息； Master Slave：实现高可用的一种方式，当主消息服务器宕机时，备服务器可以立即补充，以保证服务的继续。（2）客户端集群配置
ActiveMQ失效转移（failover）
介绍：允许当其中一台服务器宕机时，客户端在传输层上重新连接到其他消息服务器。 语法：failover:(uri1,...,uriN)？transportOptionstransportOptions参数说明：
randomize 默认为true，表示在URI列表中选择URI连接时是否采用随机策略 initialReconnectDelay 默认10，单位毫秒，表示第一次尝试重连之间的等待的时间（3）Broker Cluster集群配置
原理：当有两个节点A,B，节点A可以把消息同步到节点B，节点B也可以把消息同步到节点A,通过消息同步之后，节点A接收到的消息可以给节点B消费掉，同理，节点B接收到的消息可以给节点A消费掉。它的实现方式是采用的网络连接器的方式。
NetwrokConncetor（网络连接器）：网络连接器主要用于配置ActiveMQ服务器与服务器之间的网络通讯方式，用于服务器透传消息，网络连接器分为静态连接器和动态连接器。
静态连接器：即在服务器ip地址上具体指定ip地址，通过以下配置，当服务器比较多的时候，使用静态连接就比较麻烦，这个时候就需要用到动态连接了
动态连接器：使用多播的方式通知其他服务器，配置如下，我们首先需要定义网络连接器和传输连接器，传输连接器会告知我们一个发现的uri地址，这个就是我们的主播地址，这样我们就可以达到动态扩展服务器的效果了。
（4）Master/Slave集群配置
ActiveMQ Master Slave 集群方案：
Share nothing storage master/slave （已过时，5.8+后移除） Shared storage master/slave 共享存储 Replicated LevelDB Store 基于复制的LevelDB Store共享存储集群的原理：我们把节点A,B的持久化配置到同一个地方，先启动节点A，此时节点A获取到排它锁，节点A独占资源成为Master，而节点B无法获得锁资源，节点B就成了Slave；节点A就获得了对外开放服务的能力，可以通过外部的客户端提交信息到节点A，但是不能发送消息到节点B；如果此时节点A挂了，那么节点B立即获得了持久化资源的排它锁，成为新的Master，接收到外部客户端，而客户端使用失效转移之后，将消息发送到节点B，这个时候完成整个请求的不间断性，完成了高可用
基于复制的LevelDB Store的原理：因为LevelDB是基于ZK的，所以它的服务器至少需要三台，假设我们有三个服务器节点A,B,C，每个节点都有自己的储存方式，它们都配置同一个ZooKeeper节点，通过ZK来选举一台服务器作为Master，比如选举节点A作为Master，这时节点A就具有对外部提供服务的能力，而节点B,C是不具备的，节点A获取了服务器的外部消息资源后，它首先在本地储存，然后通过ZK将消息同步给B,C，然后B,C分别在自己的服务器里储存，这就是基于复制LevelDB的方式，如果节点A出现故障，那么ZK会立即选举一个新的Master出来。
（5）两种集群方式对比
Master/Slave：它可以做到高可用，当一台服务器挂了，其他服务器可以立即补充上去，并且保证了消息不会丢失，但做不了负载均衡，因为Slave不具备外部服务器提供服务的能力；
Broker Cluster：它不具备高可用的能力，因为它自己的消息并没有在一个地方储存，也就是说当一台服务器挂了的时候，它正在处理的消息可能会同步丢失，但是它可以做到负载均衡，即节点A,B上的消息可以互相消费。
（6）三台服务器的完美集群方案（既实现高可用，又实现负载均衡）
因为基于复制LevelDB的方式至少需要三台服务器，所以这里我们使用共享持久化资源方式；我们首先将节点A,B组成消息同步，然后将节点A,C也组成消息同步；节点B,C组成Master/Slave，然后按顺序启动A,B,C。这个方案能够实现节点A,B,C任意一台服务器宕机时，对整个集群不受影响，仍然可以正常工作，但是需要尽快恢复宕机服务器的问题，如果A,B同时宕机，整个集群就会崩溃了。所以如果要提高稳定性，可以增加服务器数量。
四、使用其他消息中间件
前面主要介绍了ActiveMQ，其优点是支持JMS，JMS让Java的开发变得简单，但其在各方面表现都比较中庸，也存在自己的问题，比如吞吐量没有Kafka高，稳定性没有RabbitMQ强。
1.企业开发需要解决的问题
不同业务系统分别处理同一消息，同一业务系统负载处理同类消息 解决消息发送时的一致性问题 解决消息处理时的幂等性问题 基于消息机制建立事件总线2.使用其他消息中间件时，分析需要做的事
解决各业务系统集群处理同一条消息 实现自己的消息提供者3.RabbitMQ：使用交换器绑定到队列
RabbitMQ消息提供者源码解析：
3.集成Kafka
Kafka使用group.id分组消费者
配置消费者参数group.id相同时对消息进行负载处理 配置服务器partitions参数，控制同一个group.id下的consumer数量小于partitions Kafka只保证同一个partition下的消息是有序的Kafka消息提供者源码解析：
当前文章：Java消息中间件之ActiveMQ
网页URL：http://cdiso.cn/article/cppopc.html
