消息队列之学习篇-3.29

应用解耦：在分布式系统中，不同的应用之间需要相互通信，但直接依赖会导致代码耦合度过高。使用MQ作为中间件，可以进行解耦。每个应用只需向MQ发送消息，而不需要了解其他应用的具体实现细节。
流量销峰：使用MQ作为中间件，将高峰期的请求缓存到队列中，在系统负载降低后再逐一处理。这可以避免系统崩溃或响应变慢。
异步调用：应用向MQ发送消息，并不需要等待消息的处理结果，从而实现异步调用。处理结果可以通过回调函数或其他方式返回给应用。
削峰填谷：使用MQ作为缓存，将请求写入队列中并按照一定速率消费，避免突发请求对系统造成压力过大。当系统负载较低时，可以加速消费消息以“填谷”，保证在高峰期前消费完所有消息。

二、常见的MQ对比

常见的MQ有如下四种：

ActiveMQ、RabbitMQ、Kafka、RocketMQ

这四种消息队列（MQ）系统在设计和应用场景上存在以下区别：

ActiveMQ: 基于JMS规范的开源消息中间件，采用主从架构模式，支持多种协议传输，如TCP、UDP、SSL等，适合企业级应用。
RabbitMQ: 基于AMQP协议的开源消息队列系统，采用分布式架构模式，性能稳定，可靠性高，适合大规模部署和复杂路由需求。
Kafka: 高吞吐量、低延迟的分布式发布订阅消息系统，采用分布式架构模式，适合大数据处理场景，如日志收集、流处理等。
RocketMQ: 阿里巴巴开源的分布式消息队列系统，采用分布式架构模式，具有高可用、高性能、高可靠性等特点，适合海量数据存储和实时消息处理。

在选择MQ系统时，需要考虑应用场景、需求和预算等因素。例如，如果应用场景需要高可靠性和复杂路由，可以选择RabbitMQ；如果需要处理海量数据，可以选择Kafka；如果需要企业级应用并且预算不是很紧张，可以选择ActiveMQ或RocketMQ。同时还需要考虑MQ的性能、安全、易用性和社区支持等方面。

三、MQ之RocketMQ

3.1、RocketMQ核心组成部分

RocketMQ是阿里巴巴开源的分布式消息传递系统，其核心组成部分包括：

Name Server：可以看作是RocketMQ的注册中心，负责管理Broker的状态信息和路由信息，并提供动态服务发现功能。
Broker：消息的存储和传输节点，包括Master Broker和Slave Broker两种类型。Broker是RocketMQ的核心模块，负责接收并存储消息，同时提供Push/Pull接口来将消息发送给Consumer。Broker同时提供消息查询的功能，可以通过MessageID和MessageKey来查询消息。Borker会将自己的Topic配置信息实时同步到NameServer

push：推模式：消息到达消息服务器之后，主动推送给消费者

pull：拉模式：是消费端发起请求，主动向消息服务器(Broker)拉取消息
Producer：消息生产者，将消息发送到Broker中心。
Consumer：消息消费者，从Broker中心获取消息并进行处理。
Topic：消息主题，用于对消息进行分类和归纳。
Message Queue：消息队列，用于存储和分发消息。Topic和Queue是一对多的关系，一个Topic下可以包含多个Queue，主要用于负载均衡，Queue数量设置建议不要比消费者数少。发送消息时，用户只指定Topic，Producer会根据Topic的路由信息选择具体发到哪个Queue上。Consumer订阅消息时，会根据负载均衡策略决定订阅哪些Queue的消息。
Message：消息实体，包含消息体和一些元数据信息，如消息ID、消息标签等。

以上这些组件共同构成了RocketMQ的基本架构，使得RocketMQ具备了高可用性、高性能、可扩展性等特点。

3.2、RocketMQ工作过程

RocketMQ的消息发送和消费过程如下：

Producer将消息发送到Broker，同时记录发送状态。
Broker收到消息后，将消息存储到对应的Topic中，并返回响应消息给Producer。
消费者向Broker订阅Topic中的消息。
Broker根据订阅关系将消息推送给对应的Consumer。
Consumer接收到消息后进行处理。
Consumer处理完消息后向Broker发送确认消息。
Broker收到确认消息后更新消息的消费状态。
如果有其他Consumer也订阅了该消息，则重复步骤5-8直至所有Consumer都消费完该消息。
消息在Broker上存储一段时间后会被删除。

Broker：消息的存储和传输节点包括Master Broker（主代理）和Slave Broker（从代理），它们的功能有所不同。

主代理是负责管理整个系统的中心节点，它接收并处理来自客户端和从代理的请求，协调各个部分之间的通信，并维护系统的状态。主代理还可以进行读写操作，以及执行其他需要全局知识的任务。

从代理则是主代理的辅助节点，它们通过与主代理通信来共同管理系统。从代理通常不会执行写操作，而只是响应主代理发出的请求，并返回数据。从代理还可以帮助分担系统负载，提高系统性能和可扩展性。

因此，Master Broker和Slave Broker在功能上存在明显的差异，主要体现在主代理具有更高级别的管理和协调职责，而从代理则主要用于提供数据服务和辅助管理。

3.3、RocketMQ消费模式

RocketMQ的消费模式主要包括两种：集群消费和广播消费。

1.集群消费（Clustered Consumer）：

集群消费是指多个消费者共同消费一个主题下的消息，每个消费者只处理部分消息子集。RocketMQ会将主题下的消息按照一定的规则均匀地分配到各个消费者上，以实现负载均衡。这样可以提高系统吞吐量，但会导致同一消息被多个消费者重复消费，因此不适用于需要去重的场景。

2.广播消费（Broadcasting Consumer）：

广播消费是指每个消费者都可以消费主题下的所有消息，各个消费者之间互相独立，互不影响。适用于需要对所有消息进行处理的场景，例如日志记录等。广播消费的缺点是会造成消息冗余消费，降低了系统性能。

需要注意的是，RocketMQ还提供了顺序消息的消费模式，这是通过设置MessageQueueSelector来实现的，具体实现方式与上述两种模式有所不同。

3.4、RocketMQ消费模式

RocketMQ的消息类型主要分为普通消息和顺序消息两种。

1.普通消息（Normal Message）：

普通消息是RocketMQ中最常用的消息类型，它没有特殊的约束条件，可以按照任意顺序发送和消费。普通消息一旦被发送到Broker，就会立即被存储，等待被消费者消费。普通消息不保证消息的顺序和唯一性，适合于实时性要求不高的场景。

2.顺序消息（Orderly Message）：

顺序消息是指在同一个消息队列中的消息必须严格按照发送顺序被消费。顺序消息可以保证消息的有序性，但是需要注意的是，在使用顺序消息时，消息生产者需要保证对同一个消息队列的顺序写入，而消费者需要对同一个消息队列进行顺序消费。如若违反此规则，可能会导致消息的顺序错乱。

此外，RocketMQ还支持延迟消息、事务消息和批量消息等类型。延迟消息允许将消息发送到Broker之后延迟一定时间再被消费者消费，用于实现各种延迟任务。事务消息则需要在消息发送前开启事务，并在确认消息发送成功后提交事务，以确保消息的安全性。批量消息可以将多条消息合并到一起发送，以提高系统吞吐量。

3.5、RocketMQ集群搭建

RoketMQ集群搭建

RocketMQ

分布式集群是通过Master和Slave的配合达到高可用性的。

RocketMQ集群有两种模式可供搭建：

Master-Slave模式：一个Master节点对应多个Slave节点，Master节点处理写请求并将数据同步到所有的Slave节点上，读取时可以从任意一个节点读取。这种模式实现了高可用和负载均衡。
Broker-Cluster模式：多个Broker节点组成一个Cluster，每个节点都可以独立地处理读写请求。这种模式实现了高吞吐量和可扩展性，但不具备高可用性，需要通过自定义路由策略来实现负载均衡。

3.6、RocketMQ可视化平台安装

安装可视化平台RocketMq-Dashboard

在下面地址：

mirrors / apache / rocketmq-dashboard · GitCode

下载，用IDEA打开，在.yml文件修改服务器地址为项目中rocketMQ服务器地址,然后启动项目,在浏览器输入:

http://localhost:8080/

即可访问可视化界面。

四、MQ之RabbitMQ

4.1、RabbitMQ核心概念

RabbitMQ四大核心概念：生产者、交换机、消费者、队列。

生产者：

产生数据发送消息的程序是生产者.

交换机：

交换机是 RabbitMQ 非常重要的一个部件，一方面它接收来自生产者的消息，另一方面它将消息推送到队列中。交换机必须确切知道如何处理它接收到的消息，是将这些消息推送到特定队列还是推送到多个队列，亦或者是把消息丢弃，这个得有交换机类型决定.常见的交换机有如下四种类型：

1.direct Exchange(直接交换机)

匹配路由键，只有完全匹配消息才会被转发

2.Fanout Excange（广播交换机）

将消息发送至所有的队列

3.Topic Exchange(主题交换机)

将路由按模式匹配，此时队列需要绑定要一个模式上。符号“#”匹配一个或多个词，符号“匹配不多不少一个词。因此“abc.#”能够匹配到“abc.def.ghi”，但是“abc.” 只会匹配到“abc.def”。

4.Header Exchange

在绑定Exchange和Queue的时候指定一组键值对，header为键，根据请求消息中携带的header进行路由

队列：

队列是 RabbitMQ 内部使用的一种数据结构，尽管消息流经 RabbitMQ 和应用程序，但它们只能存储在队列中。队列仅受主机的内存和磁盘限制的约束，本质上是一个大的消息缓冲区。许多生产者可以将消息发送到一个队列，许多消费者可以尝试从一个队列接收数据。这就是我们使用队列的方式.

消费者：

消费与接收具有相似的含义。消费者大多时候是一个等待接收消息的程序。请注意生产者，消费者和消息中间件很多时候并不在同一机器上。同一个应用程序既可以是生产者又是可以是消费者。

4.2、RabbitMQ的工作原理

RabbitMQ是一个消息中间件，它通过队列来存储和转发消息。RabbitMQ的工作原理如下：

生产者将消息发送到交换机（Exchange）上。
交换机根据路由键（Routing Key）将消息发送到相应的队列（Queue）中。
消费者从队列中接收消息进行处理。

RabbitMQ支持多种交换机类型，包括Direct、Topic、Fanout和Headers等，不同类型的交换机有不同的路由规则。同时，RabbitMQ还支持消息确认机制和消息持久化机制，确保消息传递的可靠性和稳定性。

Broker：接收和分发消息的应用，RabbitMQ Server 就是 Message Broker

Virtual host：出于多租户和安全因素设计的，把 AMQP 的基本组件划分到一个虚拟的分组中，类似于网络中的 namespace 概念。当多个不同的用户使用同一个RabbitMQ server 提供的服务时，可以划分出多个 vhost，每个用户在自己的 vhost 创建 exchange／queue 等

Connection：publisher／consumer 和 broker 之间的 TCP 连接

Channel：如果每一次访问 RabbitMQ 都建立一个 Connection，在消息量大的时候建立 TCP Connection 的开销将是巨大的，效率也较低。Channel 是在 connection 内部建立的逻辑连接，如果应用程序支持多线程，通常每个 thread 创建单独的 channel 进行通讯，AMQP method 包含了 channel id 帮助客户端和 message broker 识channel，所以 channel 之间是完全隔离的。Channel 作为轻量级的Connection 极大减少了操作系统建立 TCP connection 的开销

Exchange：message 到达 broker 的第一站，根据分发规则，匹配查询表中的 routing key，分发消息到 queue 中去。常用的类型有：direct (point-to-point), topic (publish-subscribe) and fanout (multicast)

Queue：消息最终被送到这里等待 consumer 取走

Binding：exchange 和 queue 之间的虚拟连接，binding 中可以包含 routing key，Binding 信息被保存到 exchange 中的查询表中，用于 message 的分发依据

五、Kafka

5.1、基本定义

传统定义：Kafka是一个分布式的基于发布/订阅模式的消息队列(Message Queue)，主要应用于大数据实时处理领域。

最新定义：一个开源的分布式事件流平台，可应用于高性能数据通道、流分析、数据集成、关键任务应用。

5.2、Kafka工作原理

Producer：消息⽣产者，向 Kafka Broker 发消息的客户端。
Consumer：消息消费者，从 Kafka Broker 取消息的客户端。Kafka支持持久化，生产者退出后，未消费的消息仍可被消费。
Consumer Group：消费者组（CG），消费者组内每个消费者负责消费不同分区的数据，提⾼消费能⼒。⼀个分区只能由组内⼀个消费者消费，消费者组之间互不影响。所有的消费者都属于某个消费者组，即消费者组是逻辑上的⼀个订阅者。
Broker：⼀台 Kafka 机器就是⼀个 Broker。⼀个集群(kafka cluster)由多个 Broker 组成。⼀个 Broker 可以容纳多个 Topic。
Controller：由zookeeper选举其中一个Broker产生。它的主要作用是在 Apache ZooKeeper 的帮助下管理和协调整个 Kafka 集群。
Topic：可以理解为⼀个队列，Topic 将消息分类，⽣产者和消费者⾯向的是同⼀个 Topic。
Partition：为了实现扩展性，提⾼并发能⼒，⼀个⾮常⼤的 Topic 可以分布到多个 Broker上，⼀个 Topic 可以分为多个 Partition，同⼀个topic在不同的分区的数据是不重复的，每个 Partition 是⼀个有序的队列，其表现形式就是⼀个⼀个的⽂件夹。不同Partition可以部署在同一台机器上，但不建议这么做。
Replication：每⼀个分区都有多个副本，副本的作⽤是做备胎。当主分区（Leader）故障的时候会选择⼀个备胎（Follower）上位，成为Leader。在kafka中默认副本的最⼤数量是10个，且副本的数量不能⼤于Broker的数量，follower和leader绝对是在不同的机器，同⼀机器对同⼀个分区也只可能存放⼀个副本（包括⾃⼰）。
Message：每⼀条发送的消息主体。
Leader：每个分区多个副本的“主”副本，⽣产者发送数据的对象，以及消费者消费数据的对象，都是 Leader。
Follower：每个分区多个副本的“从”副本，使用发布订阅模式主动拉取Leader的数据（与redis不同），实时从 Leader 中同步数据，保持和 Leader 数据的同步。Leader 发⽣故障时，某个 Follower 还会成为新的 Leader。
Offset：消费者消费的位置信息，监控数据消费到什么位置，当消费者挂掉再重新恢复的时候，可以从消费位置继续消费。
ZooKeeper：Kafka 集群能够正常⼯作，需要依赖于 ZooKeeper，ZooKeeper 帮助 Kafka存储和管理集群信息。
High Level API 和Low Level API ：高水平API，kafka本身定义的行为，屏蔽细节管理，使用方便；低水平API细节需要自己处理，较为灵活但是复杂。

kafka 存储的消息来⾃任意多被称为 Producer ⽣产者的进程。数据从⽽可以被发布到不同的Topic 主题下的不同 Partition 分区。在⼀个分区内，这些消息被索引并连同时间戳存储在⼀起。其它被称为 Consumer 消费者的进程可以从分区订阅消息。

Kafka 运⾏在⼀个由⼀台或多台服务器组成的集群上，并且分区可以跨集群结点分布。

Kafka核心设计理念是基于发布-订阅（Publish-Subscribe）模式的消息队列。Kafka的工作原理如下：

数据的生产者向Kafka集群中的Broker发送消息。
消息通过主题（Topic）进行分类存储在Kafka中的不同分区（Partition）上。
分区中的每一条消息都有一个唯一的偏移量（Offset），消费者通过指定偏移量来消费消息。
消费者从Broker拉取数据进行处理，消费完成后将偏移量提交到服务器。
Kafka支持多副本机制，保证消息的高可用性和容错性。
Kafka还支持批量读写和压缩等技术，提高了吞吐量和传输效率。

总之，Kafka具备高吞吐量、低延迟、高可靠性和可扩展性等特点，广泛应用于大规模数据流处理和实时数据管道等场景。

六、ZooKeeper

6.1、基本概念

ZooKeeper 是一个分布式协调服务，可以用于构建高可用、高性能的分布式系统。它提供了一个简单的文件系统接口和一些基本的原语，如锁、信号量和队列等，使得应用程序可以在分布式环境中进行协作，并且能够处理诸如选主、配置管理、命名服务、分布式同步等问题。Kafka的运行依赖ZooKeeper。

ZooKeeper 主要有以下几个特点：

分布式协调：ZooKeeper 的核心是一个分布式协调服务，它可以维护大规模集群中各个节点之间的状态信息，并且确保这些信息的一致性和可用性。
高性能：由于 ZooKeeper 的设计目标是支撑高并发、低延迟的应用场景，因此它采用了高性能的内存数据库来存储数据，并使用了多种优化技术，如快照和事务日志等，以确保数据的可靠性和性能的高效性。
可扩展性：ZooKeeper 支持水平扩展，可以通过增加节点来提高系统的容量和性能，并且在节点动态变化时能够自动完成数据重新平衡。
简单易用：ZooKeeper 提供了简单易用的 API 和命令行工具，使得开发人员可以轻松地进行开发和调试，同时也方便了系统管理员进行管理和维护。

ZooKeeper 已经被广泛应用于分布式领域的各种场景，如 Hadoop、HBase、Kafka 等，并成为了构建大型分布式系统的基础组件之一。

6.2、Zookeeper与Kafka的关系

ZooKeeper主要用来协调Kafka的各个broker，不仅可以实现broker的负载均衡，而且当增加了broker或者某个broker故障了，ZooKeeper将会通知生产者和消费者，这样可以保证整个系统正常运转。

我们看一下Zookeeper在Kafka工作流程中的角色,Kafka流程如下：

1）生产者定期向主题发送消息。

2）Kafka broker将所有消息存储在为该特定主题配置的分区中。它确保消息在分区之间平等共享。如果生产者发送两个消息，并且有两个分区，则Kafka将在第一个分区中存储一个消息，在第二个分区中存储第二个消息。

3）消费者订阅一个特定的主题。

4）一旦消费者订阅了一个主题，Kafka将向消费者提供该主题的当前偏移量，并将偏移量保存在ZooKeeper中。

5）消费者将定期请求Kafka新消息。

6）一旦Kafka收到来自生产者的消息，它会将这些消息转发给消费者。

7）消费者将收到消息并处理它。

8）一旦消息被处理，消费者将向Kafka broker发送确认。

9）一旦Kafka收到确认，它会将偏移量更改为新值，并在ZooKeeper中进行更新。由于ZooKeeper中保留了偏移量，因此即使在服务器出现故障时，消费者也可以正确读取下一条消息。

原文链接：https://blog.csdn.net/nuist_NJUPT/article/details/129814839

一、消息队列(MQ)概述