MQ应用详解

分布式消息队列的应用思考点

生产端的消息可靠性投递就是有些业务场景我们需要消息是百分之百投递成功的，或者与我们的数据库一定是一个原子性的操作
消费端幂等在生产端为了保证消息投递可靠的时候，可能会出现重复发送消息的情况，我们的消费端一定要做好消费的幂等，杜绝出现消息重复消费的情况
MQ的高可用我们要保证MQ的节点在挂掉一个或多个，MQ还是可用的状态
MQ的低延迟在流量非常大的时候，我们如何保证消息的低延迟
MQ的消息的可靠性就是如果我们的消息落到了MQ中，如何保证消息肯定不会丢失，比如某个磁盘出现问题，还能使消息不丢失（一般都是使用分片、副本的概念解决）
MQ消息的堆积能力当消息量非常大，消费者消费速度跟不上的时候，我们的MQ能否堆积一个很大的消息量
扩展性等

主流的分布式消息队列

目前业界主流的消息中间件有： ActiveMQ、RabbitMQ、RocketMQ、Kafka

如何进行技术选型

各个MQ的性能、优缺点、响应的业务场景、
集群架构模式，分布式、可扩展性、高可用、可维护性
综合成本问题，集群规模，人员成本（既看公司的技术栈，公司整体比较熟悉哪种MQ的使用等等的综合考虑）
未来的方向、规划、思考

JMS及其专业术语

JMS（Java Message Service）规范，也就是Java消息服务，它定义了Java中访问消息中间件的接口的规范。在这里注意，JMS只是接口，并没有给予实现，实现JMS接口的消息中间件称为 “JMS Provider”，目前知名的开源 MOM （Message Oriented Middleware，也就是消息中间件）系统包括Apache的ActiveMQ、RocketMQ、Kafka，以及RabbitMQ，可以说他们都 “基本遵循” 或 “参考” JMS规范，都有自己的特点和优势

专业术语：

JMS（Java Message Service）：实现JMS 接口的消息中间件；
Provider（MessageProvider）：消息的生产者；
Consumer（MessageConsumer）：消息的消费者；
PTP（Point to Point）：即点对点的消息模型，这也是非常经典的模型；
Pub / Sub（Publish/Subscribe）：，即发布/订阅的消息模型；
Queue：队列目标，也就是我们常说的消息队列，一般都是会真正的进行物理存储；
Topic：主题目标；
ConnectionFactory：连接工厂，JMS 用它创建连接；
Connection：JMS 客户端到JMS Provider 的连接；
Destination：消息的目的地；
Session：会话，一个发送或接收消息的线程（这里Session可以类比Mybatis的Session）；

ActiveMQ

ActiveMQ介绍

ActiveMQ 是一个完全支持JMS1.1和J2EE 1.4规范的 JMS Provider实现，尽管JMS规范出台已经是很久的事情了，但是JMS在早些年的 “J2EE应用” 时期扮演着特殊的地位，可以说那个年代ActiveMQ在业界应用最广泛，当然如果现在想要有更强大的性能和海量数据处理能力，ActiveMQ还需要不断的升级版本，不断的提升性能和架构设计的重构。

就算现在我们 80% 以上的业务我们使用ActiveMQ已经足够满足需求，其丰富的API、多种集群构建模式使得他成为业界老牌消息中间件，在中小型企业中应用广泛！

当然如果你想针对大规模、高并发应用服务做消息中间件技术选型，譬如淘宝、京东这种大型的电商网站，尤其是双11这种特殊时间，ActiveMQ可能就显得力不从心了

ActiveMQ消息投递模式

点对点：生产者向队列投递一条消息，只有一个消费者能够监听得到这条消息（PTP)
发布订阅：生产者向队列投递一条消息，所有监听该队列的消费者都能够监听得到这条消息（P/S)

ActiveMQ各项指标

衡量一个MOM，我们主要从三方面考虑即可，即服务性能、存储堆积能力、可扩展性。

服务性能：ActiveMQ的性能一般，在早期传统行业为王的时代还是比较流行的，但现如今面对高并发、大数据的业务场景，往往力不从心！
数据存储：默认采用kahadb存储（索引文件形式存储），也可以使用高性能的google leveldb（内存数据库存储），或者可以使用MySql、Oracle进程消息存储（关系型数据库存储）。
集群架构：ActiveMQ 可以与zookeeper进行构建主备集群模型，并且多套的主备模型直接可以采用Network的方式构建分布式集群。

ActiveMQ集群架构模式

ActiveMQ最经典的两种集群架构模式，Master-Slave 、Network 集群模式

Master-Slave: 顾名思义，就是主从方式，当然这里要理解为主备的方式，也就是双机热备机制；Master Slave 背后的想法是，消息被复制到slave broker，因此即使master broker遇到了像硬件故障之类的错误，你也可以立即切换到slave broker而不丢失任何消息。 Master Slave是目前ActiveMQ推荐的高可靠性和容错的解决方案。

zookeeper的作用就是为了当绿色的主节点宕机时，进行及时切换到备份的灰色节点上去，使其进行主从角色的互换，用于实现高可用性的方案。
Master-Slave集群模型的缺点也显而易见，就是不能做到分布式的topic、queue，当消息量巨大时，我们的MQ集群压力过大，没办法满足分布式的需求

Network：这里可以理解为网络通信方式，也可以说叫Network of brokers。这种方式真正解决了分布式消息存储和故障转移、broker切换的问题。可以理解消息会进行均衡；从ActiveMQ1.1版本起，ActiveMQ支持networks of brokers。它支持分布式的queues和topics。一个broker会相同对待所有的订阅（subscription）：不管他们是来自本地的客户连接，还是来自远程broker，它都会递送有关的消息拷贝到每个订阅。远程broker得到这个消息拷贝后，会依次把它递送到其内部的本地连接上。（说白了就是部署多套MQ集群，以每个集群为单位进行通信，每个集群有自己的主从节点，有自己的zookeeper节点）

Network集群模型的关键点：这种方案需要两套或多套（Master-Slave）的集群模型才可以搞定，部署非常麻烦，需要两套或多套集群直接相互交叉配置，相互间能够感知到彼此的存在
Network虽然解决了分布式消息队列这个难题，但是还有很多潜在的问题，最典型的就是资源浪费问题，并且也可能达不到所预期的效果；通常采用Master-Slave模型是传统型互联网公司的首选，作为互联网公司往往会选择开箱即用的消息中间件，从运维、部署、使用各个方面都要优于ActiveMQ，当然ActiveMQ毕竟是 “老牌传统强Q”，Apache的顶级项目之一，目前正在进行新版本的重构（对于5.X版本）与落地。

RibbitMQ

RibbitMQ四种集群模式

主备模式

warren（兔子窝）：经典的主备模式，正常情况由主节点提供服务，从节点只是备份数据，当主节点挂掉，从节点会代替主节点提供服务。与Active不同的是，主从实现不是通过zookeeper来实现的，它使用Haproxy实现的，Haproxy跟Nginx有点类似

远程模式

早起版本提供的一种多活的模式，主要是服务异地的容灾，与上述ActiveMQ的NetWork模式非常类似，主要是在不同的地方部署不同的集群，可以提高容灾能力及处理性能，现在的版本已经不推荐使用

远距离通信和复制，可以实现多活的一种模式，简称Shovel模式，就是我们可以把消息进行不同的数据中心的复制工作，可以让跨地域的两个MQ集群互联，当其中的某一个集群处理消息处理不过来的时候，可以把消息转发到另外一个集群进行处理。集群之间的通信使用MQ的amqp协议来做通信的。此模式的配置非常麻烦，现在已经很少使用，只做了解就行。

镜像模式

业界使用最为广泛的模型，非常经典的Mirror镜像模式，保证数据100%的不丢失，镜像模式其实就是数据的备份。可靠性非常高，因为数据发过来之后，它需要将数据同步到MQ镜像集群中所有的节点，所有节点都会对数据做备份存储，它的模型跟ES的很像，但是它的副本是在所有的节点上。但是缺点也很明显，就是每个节点都存储了所有的数据，如果我们要扩容的话只能增加所有节点的磁盘大小，而不能通过增加机器来实现

多活模型

这种模式也是实现异地数据复制的主流模式，因为Shovel模式配置比较复杂，所以一般实现异地集群都是使用这种多活模型。这种模型需要依赖RabbitMQ的federation插件，可以实现持续可靠的AMQP数据通信，配置与应用很简单。

部署架构采用多中心模式，在两套或多套数据中心中各部署一套Rabbit集群，各中心的RabbitMQ服务除了为业务提供正常消息服务外，中心之间需要实现部分队列消息共享。集群可以是不同RibbitMQ版本的集群

RocketMQ

概述

RocketMQ是一款分布式、队列模型的消息中间件，由阿里巴巴自主研发的一款适用于高并发、高可靠性、海量数据场景的消息中间件。早期开源2.x版本名为MetaQ；15年迭代3.x版本，更名为RocketMQ，16年开始贡献到Apache，经过1年多的孵化，最终成为Apache顶级的开源项目，更新非常频繁，社区活跃度也非常高；RocketMQ参考借鉴了优秀的开源消息中间件Apache Kafka，其消息的路由、存储、集群划分都借鉴了Kafka优秀的设计思路，并结合自身的 “双十一” 场景进行了合理的扩展和API丰富。

优秀的能力与支持

支持集群模型、负载均衡、水平扩展能力
亿级别的消息堆积能力
采用零拷贝的原理、顺序写盘、随机读（索引文件）
丰富的API使用
代码优秀，底层通信框架采用Netty NIO框架
NameServer 代替 Zookeeper
强调集群无单点，可扩展，任意一点高可用，水平可扩展
消息失败重试机制、消息可查询
开源社区活跃度、是否足够成熟（经过双十一考验）

专业术语

Producer：消息生产者，负责产生消息，一般由业务系统负责产生消息。
Consumer：消息消费者，负责消费消息，一般是后台系统负责异步消费。
Push Consumer：Consumer的一种，需要向Consumer对象注册监听。
Pull Consumer：Consumer的一种，需要主动请求Broker拉取消息。
Producer Group：生产者集合，一般用于发送一类消息。
Consumer Group：消费者集合，一般用于接受一类消息进行消费。
Broker ： MQ消息服务（中转角色，用于消息存储与生产消费转发）

集群架构模型

RocketMQ为我们提供了丰富的集群架构模型，包括单点模式、主从模式、双主模式、以及生产上使用最多的双主双从模式（或者说多主多从模式）

Producer集群就是生产者集群（他们在同一个生产者组 Producer Group）
Consumer集群就是消费者集群（他们在同一个消费者组 Consumer Group）
NameServer集群作为超轻量级的配置中心，只做集群元数据存储和心跳工作，不必保障节点间数据强一致性，也就是说NameServer集群是一个多机热备的概念。
对于Broker而言，通常Master与Slave为一组服务，他们互为主从节点，通过NameServer与外部的Client端暴露统一的集群入口。Broker就是消息存储的核心MQ服务了。

RocketMQ作为国内顶级的消息中间件，其性能主要依赖于天然的分布式Topic/Queue，并且其内存与磁盘都会存储消息数据，借鉴了Kafka的 “空中接力” 概念，所谓 “空中接力” 就是指数据不一定要落地，RocketMQ提供了同步/异步双写、同步/异步复制的特性。在真正的生产环境中应该选择符合自己业务的配置。下面针对于RocketMQ的高性能及其瓶颈在这里加以说明：

RocketMQ目前其主要瓶颈最终会落在IOPS上面，当高峰期来临的时候，磁盘读写能力是主要的性能瓶颈，为什么瓶颈在IOPS? 根本原因还是因为云环境导致的问题，云环境的SSD物理存储显然和自建机房SSD会有不小的差距，这一点我们无论是从数据库的磁盘性能、还是搜索服务（ElasticSearch）的磁盘性能，都能给出准确的瓶颈点，单机IOPS达到1万左右就是云存储SSD的性能瓶颈，这个也解释了 “木桶短板原理” 的效应，在真正的生产中，CPU的工作主要在等待IO操作，高并发下 CPU资源接近极限，但是IOPS还是达不到我们想要的效果。

与KAFKA对比

既然RocketMQ有Kafka所有的优点，那么它两的区别在哪呢？

消息投递实时性

Kafka使用短轮询方式，实时性取决于轮询间隔时间
RocketMQ使用长轮询，同Push方式实时性一致，消息的投递延时通常在几个毫秒。

严格的消息顺序

Kafka支持消息顺序，但是一台Broker宕机后，就会产生消息乱序
RocketMQ支持严格的消息顺序，在顺序消息场景下，一台Broker宕机后，发送消息会失败，但是不会乱序

namesrv VS zk

kafka和rocketMq在协调节点选择上的差异，kafka通过zookeeper来进行协调，而rocketMq通过自身的namesrv进行协调。
kafka在具备选举功能，在Kafka里面，Master/Slave的选举，有2步：第1步，先通过ZK在所有机器中，选举出一个KafkaController；第2步，再由这个Controller，决定每个partition的Master是谁，Slave是谁。因为有了选举功能，所以kafka某个partition的master挂了，该partition对应的某个slave会升级为主对外提供服务。
rocketMQ不具备选举，Master/Slave的角色也是固定的。当一个Master挂了之后，你可以写到其他Master上，但不能让一个Slave切换成Master。那么rocketMq是如何实现高可用的呢，其实很简单，rocketMq的所有broker节点的角色都是一样，上面分配的topic和对应的queue的数量也是一样的，Mq只能保证当一个broker挂了，把原本写到这个broker的请求迁移到其他broker上面，而并不是这个broker对应的slave升级为主。

吞吐量

kafka在消息存储过程中会根据topic和partition的数量创建物理文件，也就是说我们创建一个topic并指定了3个partition，那么就会有3个物理文件目录，也就说说partition的数量和对应的物理文件是一一对应的。
rocketMq在消息存储方式就一个物流问题，也就说传说中的commitLog，rocketMq的queue的数量其实是在consumeQueue里面体现的，在真正存储消息的commitLog其实就只有一个物理文件。
kafka的多文件并发写入 VS rocketMq的单文件写入，性能差异kafka完胜可想而知。
kafka的大量文件存储会导致一个问题，也就说在partition特别多的时候，磁盘的访问会发生很大的瓶颈，毕竟单个文件看着是append操作，但是多个文件之间必然会导致磁盘的寻道。

在性能上Kafka是完胜的

KAFKA

介绍

Kafka是LinkedIn开源的分布式消息系统，目前归属于Apache顶级项目

主要特点是基于Pull的模式来处理消息消费，追求高吞吐量，一开始就是用于日志收集

0.8版本开始支持复制，不支持事务，对消息的重复、丢失、错误没有严格要求，适合产生大量数据的互联网服务的数据收集业务

特点

有分布式的特性，就是支持分区的概念，一个主题下可以有多个分区
有跨平台的特性，支持不同语言的客户端
堆积能力特别强，且并不影响消息的接收和发送
实时性非常强

高性能的原因（重点）

顺序写

就是顺序写盘，可以提高磁盘的利用率，就是一个一个的写，而不是随机写，这样会大大提高写的性能。

每个topic有不同的分区，而每个分区下包含若干个只能追加写的提交日志：新消息被追加到文件的最末端。最直接的证明就是Kafka源码中只调用了FileChannel.write(ByteBuffer)，而没有调用过带offset参数的write方法，说明它不会执行随机写操作。

Page Cache

首先，Kafka重度依赖底层操作系统提供的PageCache功能。当上层有写操作时，操作系统只是将数据写入PageCache，同时标记Page属性为Dirty。当读操作发生时，先从PageCache中查找，如果发生缺页才进行磁盘调度，最终返回需要的数据。实际上PageCache是把尽可能多的空闲内存都当做了磁盘缓存来使用。同时如果有其他进程申请内存，回收PageCache的代价又很小，所以现代的OS都支持PageCache。

使用PageCache功能同时可以避免在JVM内部缓存数据，JVM为我们提供了强大的GC能力，同时也引入了一些问题不适用与Kafka的设计。

如果在Heap内管理缓存，JVM的GC线程会频繁扫描Heap空间，带来不必要的开销。如果Heap过大，执行一次Full GC对系统的可用性来说将是极大的挑战。
所有在在JVM内的对象都不免带有一个Object Overhead(千万不可小视)，内存的有效空间利用率会因此降低。
所有的In-Process Cache在OS中都有一份同样的PageCache。所以通过只在PageCache中做缓存至少可以提高一倍的缓存空间。
如果Kafka重启，所有的In-Process Cache都会失效，而OS管理的PageCache依然可以继续使用。

零拷贝

首先介绍一下传统的网络I/O操作流程，大体上分为以下4步：

OS从硬盘把数据读到内核区的PageCache。
用户进程把数据从内核区Copy到用户区。
然后用户进程再把数据写入到Socket，数据流入内核区的Socket Buffer上。
OS再把数据从Buffer中Copy到网卡的Buffer上，这样完成一次发送。

整个过程一共经历了四次拷贝，同一份数据在内核Buffer与用户Buffer之间重复拷贝，效率低下。其中2、3两步没有必要，完全可以直接在内核区完成数据拷贝。

零拷贝技术就是省略了第2、3步，不难看出，Kafka的设计初衷是尽一切努力在内存中完成数据交换，无论是对外作为一整个消息系统，或是内部同底层操作系统的交互。如果Producer和Consumer之间生产和消费进度上配合得当，完全可以实现数据交换零I/O。这也就是我为什么说Kafka使用“硬盘”并没有带来过多性能损失的原因。

主要特点

同时为发布和订阅提供高吞吐量。据了解，Kafka每秒可以生产约25万消息（50 MB），每秒处理55万消息（110 MB）。
可进行持久化操作。将消息持久化到磁盘，因此可用于批量消费，例如ETL，以及实时应用程序。通过将数据持久化到硬盘以及replication防止数据丢失。
分布式系统，易于向外扩展。所有的producer、broker和consumer都会有多个，均为分布式的。无需停机即可扩展机器。
消息被处理的状态是在consumer端维护，而不是由server端维护。当失败时能自动平衡。
支持online和offline的场景。

Kafka的架构

Kafka的整体架构非常简单，是显式分布式架构，producer、broker（kafka）和consumer都可以有多个。Producer，consumer实现Kafka注册的接口，数据从producer发送到broker，broker承担一个中间缓存和分发的作用。broker分发注册到系统中的consumer。broker的作用类似于缓存，即活跃的数据和离线处理系统之间的缓存。客户端和服务器端的通信，是基于简单，高性能，且与编程语言无关的TCP协议。

基本概念

Topic：特指Kafka处理的消息源（feeds of messages）的不同分类。
Partition：Topic物理上的分组，一个topic可以分为多个partition，每个partition是一个有序的队列。partition中的每条消息都会被分配一个有序的id（offset）。
Message：消息，是通信的基本单位，每个producer可以向一个topic（主题）发布一些消息。
Producers：消息和数据生产者，向Kafka的一个topic发布消息的过程叫做producers。
Consumers：消息和数据消费者，订阅topics并处理其发布的消息的过程叫做consumers。
Broker：缓存代理，Kafka集群中的一台或多台服务器统称为broker。

发送消息的流程

Producer根据指定的partition方法（round-robin、hash等），将消息发布到指定topic的partition里面
kafka集群接收到Producer发过来的消息后，将其持久化到硬盘，并保留消息指定时长（可配置），而不关注消息是否被消费。
Consumer从kafka集群pull数据，并控制获取消息的offset

kafka的优秀设计

从kafka的吞吐量、负载均衡、消息拉取、扩展性来说一说kafka的优秀设计。

高吞吐

高吞吐是kafka需要实现的核心目标之一，为此kafka做了以下一些设计：

内存访问：直接使用 linux 文件系统的cache，来高效缓存数据，对数据进行读取和写入。
数据磁盘持久化：消息不在内存中cache，直接写入到磁盘，充分利用磁盘的顺序读写性能。
zero-copy：减少IO操作步骤，采用linux Zero-Copy提高发送性能。传统的数据发送需要发送4次上下文切换，采用sendfile系统调用之后，数据直接在内核态交换，系统上下文切换减少为2次。根据测试结果，可以提高60%的数据发送性能。Zero-Copy详细的技术细节可以参考：https://www.ibm.com/developerworks/linux/library/j-zerocopy/
对消息的处理：支持数据批量发送、支持数据压缩机制
主题分区：Topic划分为多个partition，提高生产/消费端处理消息的parallelism（并行度），数据在磁盘上存取代价为O(1)。kafka以topic来进行消息管理，每个topic包含多个part（ition），每个part对应一个逻辑log，有多个segment组成。每个segment中存储多条消息，消息id由其逻辑位置决定，即从消息id可直接定位到消息的存储位置，避免id到位置的额外映射。每个part在内存中对应一个index，记录每个segment中的第一条消息偏移。发布者发到某个topic的消息会被均匀的分布到多个part上（随机或根据用户指定的回调函数进行分布），broker收到发布消息往对应part的最后一个segment上添加该消息，当某个segment上的消息条数达到配置值或消息发布时间超过阈值时，segment上的消息会被flush到磁盘，只有flush到磁盘上的消息订阅者才能订阅到，segment达到一定的大小后将不会再往该segment写数据，broker会创建新的segment。

负载均衡

producer根据用户指定的算法，将消息发送到指定的partition
存在多个partiiton，每个partition有自己的replica，每个replica分布在不同的Broker节点上
多个partition需要选取出lead partition，lead partition负责读写，并由zookeeper负责fail over
通过zookeeper管理broker与consumer的动态加入与离开

消息的拉取

简化kafka设计（由于kafka broker会持久化数据，broker没有内存压力，因此，consumer非常适合采取pull的方式消费数据）
consumer根据消费能力自主控制消息拉取速度
consumer根据自身情况自主选择消费模式，例如批量，重复消费，从尾端开始消费等

可扩展性

当需要增加broker结点时，新增的broker会向zookeeper注册，而producer及consumer会根据注册在zookeeper上的watcher感知这些变化，并及时作出调整。

KAFKA应用场景

消息队列：比起大多数的消息系统来说，Kafka有更好的吞吐量，内置的分区，冗余及容错性，这让Kafka成为了一个很好的大规模消息处理应用的解决方案。消息系统一般吞吐量相对较低，但是需要更小的端到端延时，并常常依赖于Kafka提供的强大的持久性保障。在这个领域，Kafka足以媲美传统消息系统，如ActiveMQ或RabbitMQ。
行为跟踪：Kafka的另一个应用场景是跟踪用户浏览页面、搜索及其他行为，以发布-订阅的模式实时记录到对应的topic里。那么这些结果被订阅者拿到后，就可以做进一步的实时处理，或实时监控，或放到hadoop/离线数据仓库里处理。
元信息监控：作为操作记录的监控模块来使用，即汇集记录一些操作信息，可以理解为运维性质的数据监控吧。
日志收集：日志收集方面，其实开源产品有很多，包括Scribe、Apache Flume。很多人使用Kafka代替日志聚合（log aggregation）。日志聚合一般来说是从服务器上收集日志文件，然后放到一个集中的位置（文件服务器或HDFS）进行处理。然而Kafka忽略掉文件的细节，将其更清晰地抽象成一个个日志或事件的消息流。这就让Kafka处理过程延迟更低，更容易支持多数据源和分布式数据处理。比起以日志为中心的系统比如Scribe或者Flume来说，Kafka提供同样高效的性能和因为复制导致的更高的耐用性保证，以及更低的端到端延迟。
流处理：这个场景可能比较多，也很好理解。保存收集流数据，以提供之后对接的Storm或其他流式计算框架进行处理。很多用户会将那些从原始topic来的数据进行阶段性处理，汇总，扩充或者以其他的方式转换到新的topic下再继续后面的处理。例如一个文章推荐的处理流程，可能是先从RSS数据源中抓取文章的内容，然后将其丢入一个叫做“文章”的topic中；后续操作可能是需要对这个内容进行清理，比如回复正常数据或者删除重复数据，最后再将内容匹配的结果返还给用户。这就在一个独立的topic之外，产生了一系列的实时数据处理的流程。Strom和Samza是非常著名的实现这种类型数据转换的框架。
事件源：事件源是一种应用程序设计的方式，该方式的状态转移被记录为按时间顺序排序的记录序列。Kafka可以存储大量的日志数据，这使得它成为一个对这种方式的应用来说绝佳的后台。比如动态汇总（News feed）
持久性日志（commit log）：Kafka可以为一种外部的持久性日志的分布式系统提供服务。这种日志可以在节点间备份数据，并为故障节点数据回复提供一种重新同步的机制。Kafka中日志压缩功能为这种用法提供了条件。在这种用法中，Kafka类似于Apache BookKeeper项目。

最后更新： 2020年04月26日 11:04

原始链接： https://jjw-story.github.io/2020/04/05/分布式消息队列/

赏