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 Group

3.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)?transportOptions

transportOptions参数说明:

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
文章路径:http://pwwzsj.com/article/cppopc.html