消息队列MQ汇总

有一家果汁生产企业

削峰（消息队列是一个仓库）

异步（不必等待消费者消费）

解耦（不用管消费者的事情）

最常见的一个场景是用户注册后，需要发送注册邮件和短信通知，以告知用户注册成功。

异步解耦是消息队列 MQ 的主要特点，主要目的是减少请求响应时间和解耦。

主要的使用场景就是将比较耗时而且不需要即时（同步）返回结果的操作作为消息放入消息队列。

同时，由于使用了消息队列MQ，只要保证消息格式不变，消息的发送方和接收方并不需要彼此联系，也不需要受对方的影响，即解耦合。

传统的做法如下：

改善后的做法

流量削峰也是消息队列 MQ 的常用场景，一般在秒杀或团队抢购(高并发)活动中使用广泛。

在秒杀或团队抢购活动中，由于用户请求量较大，导致流量暴增，秒杀的应用在处理如此大量的访问流
量后，下游的通知系统无法承载海量的调用量，甚至会导致系统崩溃等问题而发生漏通知的情况。为解
决这些问题，可在应用和下游通知系统之间加入消息队列 MQ。

流量削峰

秒杀处理流程如下所述：

削峰

而不会因为突发的超负荷的请求而完全奔溃

异步

消息队列提供了异步处理机制，允许把一个消息放入队列中，当时不处理，等想要处理的时候才处理

解耦

基于统一的数据的接口，生产者和消费者可以独立的扩展或者修改两边的处理过程

消息队列把数据进行持久化直到它们已经完全处理，防止数据处理过程中数据丢失

消息队列解耦了你的处理过程，可以灵活的增大消息的入队和处理

消息队列降低了进程间的耦合度，但一部分组件失效，不会影响整个系统，挂掉的进程仍然可以在恢复队后处理队列中的消息

消息队列提供的冗余机制保证了消息能被实际的处理，只要一个进程读取了该队列即可，“通过”消息预定的方式“保证”只送达一次

消息队列本来就是排序的，并且能够保证数据按照特定的顺序来处理。

大部分消息系统通过FIFO的顺序来处理

消息队列的缓冲能够高效的执行写入队列的处理而不受从队列读的预备处理的约束。

缓冲有利于控制和优化数据流经过系统的速度

通过消息队列消息被处理的频率来辅助确定表现不佳的处理过程

解决大数据各个子系统之间的数据的高性能、低延迟的不停流转

解决在线应用(消息)和离线应用(数据文件，日志)大规模的数据处理。作为一个沟通生产者produce和消费者消费Consume(处理分析)的桥梁

Apache Kafka

LinkedIn公司

Scala

日志服务

消息系统

分布式的

基于发布和订阅的

可分区的

可复制的

持久性的

高性能

跨语言

快速

可扩展的

高吞吐量的

冗余备份的

相对于ActiveMQ是一个非常轻量级的消息系统

基于hadoop的批处理系统

用于处理活跃的流式数据

它主要用于处理流式数据。

kafka中的 zookeeper 起到什么作用，可以不用zookeeper么 ？

一个分布式，支持分区(partition)，多副本的(replica)，基于zookeeper协调的分布式消息系统

Kafka用到了zookeeper，所有首先启动zookeeper服务，再启动Kafka

可以在命令的结尾加个&符号，这样就可以启动后离开控制台。【守护进程】

保存着meta数据

另外，还负责broker故障发现，partition leader选举，负载均衡等功能

zookeeper 是一个分布式的协调组件

早期版本

新版本

新的consumer使用了kafka内部的group coordination协议，也减少了对zookeeper的依赖，

但是broker依然依赖于ZK，zookeeper 在kafka中还用来选举controller 和 检测broker是否存活等等。

Apache

Confluent

Cloudera/Hortonworks

maven项目

参数配置

producer，consumer等

下载解压

启动服务

启动kafka

创建topic

通过list命令查看创建的topic

发送消息

启动consumer

Kafka，将消息以Topic主题为单位进行归纳。

将向Kafka topic发布消息的程序，称为producers.

将订阅topics，并消费消息的程序，称为consumer.

Kafka以集群的方式运行，可以由一个或多个服务组成，每个服务叫做一个broker.

 producers通过网络，将消息发送到Kafka集群，集群向消费者提供消息

架构

消息发送图

图解Kafka组件之间的关系

图解Kafka组件之间的关系

图解Kafka

分布式消息引擎

消息中间件

消息服务

消息

异步通信

峰值压力缓冲

系统之间解耦合

降低系统组网复杂度

降低编程复杂度

建立实时数据流管道 从而可靠的在系统或程序间共享数据

有更好的吞吐量，内置的分区，冗余以及容错性，这让kafka成为一个很好的大规模消息处理应用的解决方案

对于一些常规的消息系统，kafka是个不错的选择

partitons/replication和容错，可以使kafka具有良好的扩展性和性能优势

kafka并没有提供JMS中的企业级特性

kafka只能使用作为"常规"的消息系统

在一定程度上，尚未确保消息的发送与接收绝对可靠

监控数据处理

作为操作记录的监控模块来使用，即汇聚记录一些操作信息，可以理解为运维性质的数据监控

monitor 监控

性能

kafka的特性决定它非常适合作为"日志收集中心"

application可以将操作日志批量异步的发送到kafka集群中，而不是保存在本地或者DB中 

kafka可以批量提交消息/压缩消息等，这对producer端而言，几乎感觉不到性能的开支

此时consumer端，可以使hadoop等其他系统化的存储和分析系统.

文件的细节，将其更清晰地抽象成一个个日志或事件的消息流。

这让kafka处理过程延迟更低，更容易支持多数据源和分布式处理。

相对于Flume，kafka提供同样高效的性能和因为复制导致的更高的耐用性保证，以及更低端到端延迟

Kafka可以为一种外部的持久性日志的分布式系统提服务。

这种日志可以在节点间备份数据，并为故障点数据恢复提供一种重新同步的机制。

kafka中日志压缩功能为这种用法提供了条件。

在这种用法中，kafka类似于Apache BookKeeper项目

持久化队列，像日志简单的读取好追加到文件尾

存储

事件源是一种应用程序设计的方式，该方式的状态转移被记录为按时间顺序排序的记录序列。

Kafka可以存储大量的日志数据，使得它成为一个对这种方式的应用来说绝佳的后台，比如动态汇总New feed

消息数据写到磁盘和备份分区

磁盘结构很好扩展

特殊的分布式文件系统

访问日志

网站活动跟踪

跟踪用户的浏览页面，搜索及其它行为，

以发布-订阅的方式实时记录到对应的topic里，这些结果被订阅者拿到后，可进一步实时处理或监控，或放到hadoop/离线数仓中处理

各大公司一般把kafka用在用户行为日志的采集和传输上，比如大数据团队要收集APP上用户的一些日志，这种日志就是用kafka来收集和传输的，因为这种日志适当丢失数据也没有关系，而且一般量特别大，要求吞吐量高，一般就是收发消息，不需要太多高级功能。所以比较适合这种场景。

作为"网站活性跟踪"的最佳工具

可以将网页/用户操作等信息发送到kafka中

活性跟踪（Website activity tracking）

在大数据中，使用了大量的数据。 关于数据，我们有两个主要挑战。

第一个挑战是如何收集大量的数据，第二个挑战是分析收集的数据。 

为了克服这些挑战，您必须需要一个消息系统。

Kafka专为分布式高吞吐量系统而设计。 Kafka往往工作得很好，作为一个更传统的消息代理的替代品。

与其他消息传递系统相比，Kafka具有更好的吞吐量，内置分区，复制和固有的容错能力，这使得它非常适合大规模消息处理应用程序。

适用于需要高吞吐量，允许适当的消息丢失，不需要太高级功能|日志采集|标杆

支持Hadoop数据并行加载，对于像Hadoop的一样的日志数据和离线分析系统，但又要求实时处理的限制，这是一个可行的解决方案。

Kafka通过Hadoop的并行加载机制统一了在线和离线的消息处理。

kafka对于可扩展的数据持久性的支持，非常适合向hadoop或者数仓中进行数据装载

在各个App，流处理，数据ETL（Extract Transform and Load）等应用服务中，担任着“粘合剂”的角色。

分布式流平台

可以实时的处理大量的数据满足各种需求场景

构建实时流应用程序，能够变换或者对数据进行相应处理

允许应用程序充当流处理器

数据流处理

流处理实时数据流

从一个或多个主题中获取输入流，并产生一个或多个输出流，能够有效的变输入流为输出流

storm/spark流式处理引擎

流式处理场景可能比较多

因此可用于批量消费，例如ETL，以及实时应用程序。

例如ETL，以及实时应用程序。

kafka对于可扩展的数据持久性的支持，非常适合向hadoop或者数仓中进行数据装载

图解使用场景

支持分区，分布式，实时数据处理数据源并创建新的数据源，并保证容错性

高吞吐率

低延迟的实时系统

10W/s的吞吐速率

同步异步

pull&push模式

为什么需要持久化？

"插入-获取-删除"范式

持久化到磁盘

常数时间复杂度

物理存储

kafka对于可扩展的数据持久性的支持，非常适合向hadoop或者数仓中进行数据装载

【数据持久化】在6块7200转的SATA RAID-5磁盘阵列的线性写速度差不多是600MB/s，但是随即写的速度却是100k/s，差了差不多6000倍。

通过将数据持久化到硬盘以及replication，防止数据丢失。

【消息有效期】会长久保留其中的消息，以便consumer可以多次消费，当前其中很多细节可以配置

自己不丢数据，默认每隔7天清理数据

日志追加

日志段

可扩展的持久化能力

高可用

被设计为一个分布式系统，易于向外扩展

Broker、Producer、Consumer都原生自动支持分布式，自动实现负载均衡；

【完全的分布式系统】

kafka集群支持热扩展

分布式系统，易于向外扩展。

不需要改变代码、不需要调节参数。

扩展就像调大电力按钮一样简单。

性能高，基本上发送给kafka都是毫秒级的性能。可用性很高。

所有的producer、broker和consumer都会有多个，均为分布式的。

因为消息队列解耦了处理过程，所以增大消息入队和处理的频率是很容易的，只要另外增加处理过程即可。

kafka运行在一个或多个服务器集群cluster上

无需停机即可扩展机器。

当需要增加broker节点时，新增的broker会向zookeeper注册，

而producer以及consumer会根据注册在zookeeper上的watcher感知这些变化，并及时作为调整

顺序保证

【消息顺序传输】

并行性分区主题

消息状态

高并发

拉取系统

【多订阅者】

容灾核心是日志复制

消息被处理的状态是在consumer端维护，而不是由server端维护。

当失败时能自动平衡。

允许集群中节点失败

系统的一部分组件失效时，不会影响到整个系统。

消息队列降低了进程间的耦合度，所以即使一个处理消息的进程挂掉，加入队列中的消息仍然可以在系统恢复后被处理。

Apache Kafka，相对于ActiveMQ是一个非常轻量级的消息系统

除了性能非常好之外，还是一个工作良好的分布式系统。

支持集群化部署

可支持集群部署，部分机器宕机可以继续运行。

集群关系

移除broker

复制

事务 transaction

幂等 Idempotence

三种消息策略、语义 level

生产者

消费者

Broker

可靠性

可靠性保证

可靠性保障

复制

配置

消息队列可以堆积请求，所以消费端业务即使短时间死掉，也不会影响主要业务的正常进行。

可以采用一对多的方式，一个生产者发布消息，可以被多个订阅topic的服务消费到，供多个毫无关联的业务使用。

很多时候，用户不想也不需要立即处理消息。

消息队列提供了异步处理机制，允许用户把一个消息放入队列，但并不立即处理它。

想向队列中放入多少消息就放多少，然后在需要的时候再去处理它们。

通信方式

很多时候，用户不想也不需要立即处理消息。

消息队列提供了异步处理机制，允许用户把一个消息放入队列，但并不立即处理它。

想向队列中放入多少消息就放多少，然后在需要的时候再去处理它们。

提高parallelism

解耦

扩展性

在任何重要的系统中，都会有需要不同的处理时间的元素。

例如，加载一张图片比应用过滤器花费更少的时间。

消息队列，通过一个缓冲层，来帮助任务最高效率的执行———写入队列的处理会尽可能的快速。

该缓冲有助于控制和优化数据流经过系统的速度。

上游数据时有突发流量，下游可能扛不住，或者下游没有足够多的机器来保证冗余，kafka在中间可以起到一个缓冲的作用，把消息暂存在kafka中，下游服务就可以按照自己的节奏进行慢慢处理。

在访问量剧增的情况下，应用仍然需要继续发挥作用，但是这样的突发流量并不常见；

如果为以能处理这类峰值访问为标准来投入资源随时待命，无疑是巨大的浪费。

使用消息队列，能够使关键组件顶住突发的访问压力，而不会因为突发的超负荷的请求，而完全崩溃。

离线数据加载

支持online和offline的场景

支持离线数据处理和实时数据处理。

kafka支持以消息集合为单位进行批量发送，以提高push效率

数据批量发送

因此可用于批量消费

批量发送

producer一次 发送多个消息

consumer一次收一批消息

标准二进制消息格式

pagecache 和sendfile组合 在消费者只写的情况下看不到磁盘活动

sendfile和更多的zero-copy背景知识见zero-copy

produce根据用户指定的算法，将消息传递给指定的partition

存在多个partition，每个partition有自己的replica，每个replica分布在不同的broker节点上

多个partition需要选取出lead partition，lead partition负责读写，并由zookeeper负责fail over

通过zookeeper管理broker与 consumer的动态加入与离开

对kafka0.8而言，提供一个metadata API来管理broker之间的负载，对kafka0.8而言

对kafka0.7而言，主要靠zookeeper来实现负载均衡

Compress

压缩日志

zero-copy：减少IO操作步骤

每个唯一客户端group都有一个集群配置的固定限额

基于broker

限流以后降速

现在开发出不少插件来拓展kafka的功能

比如来配合storm，hadoop，flume相关的插件

Cache Filesystem 缓存文件系统

顺序写

Zero-copy零拷贝技术

Batching of Messages 批量量处理

Pull 拉模式

Kafka 服务端

特点

集群

作用

kafka集群中Broker之间的关系

Broker是消息的代理

主题 topic

优点

类别/分类/归纳/逻辑集合

作用

物理上，不同Topic的消息分开存储

逻辑上一个Topic的消息，虽然保存于一个或多个broker上但用户只需指定消息的Topic即可生产或消费数据而不必关心数据存于何处）。

每个消息记录包含一个键 一个值和时间戳

支持多用户订阅

分布式

日志分区

Topics 和Logs

图解Kafka Log Implementation

topic存储结构

消息，是通信的基本单位。

每个Producer可以向一个Topic发布一些消息

一个主题可以有多个分区

物理集合与分组

有序的队列

其他

负载均衡

如何选择分区

Producer将消息发布到它指定的topic中，并负责决定发布到哪个分区。

负责发布消息到Kafka broker。

允许应用程序发布记录流或多个kafka的topics

发布消息

发布数据到topic

配置

分区器

批量发送

压缩（GZIP或Snappy）

生产者设计

发送方式

负载均衡

负责把记录分配到分区

可以轮训进行平衡负载

也可以用语义分区算法

发送消息时根据时间属性给一个偏移量

消息和数据生产者，向Kafka的一个topic发布消息的过程叫做producers。

Producer将消息发布到它指定的topic中,并负责决定发布到哪个分区。

通常简单的由负载均衡机制随机选择分区，但也可以通过特定的分区函数选择分区。使用的更多的是第二种。

图解生产者

connector api

图解消费者设计

Kafka的两种模式：队列模式和发布-订阅模式

Consumer Group消费者组

消息和数据消费者，订阅topics并处理其发布的消息的过程叫做consumers。

消费消息

消息消费者，向Kafka broker读取消息的客户端。

允许应用程序订阅一个或多个主题，并处理执行记录接收的记录流

集群关系图

只能保证一个分区中的消息有序，不能保证多个分区中消息有序

拦截器

消费者组监控

心跳

轮询

配置

提交

独立消费者

Controller 控制器概述

Controller 控制器的选举

Controller 控制器的作用

Controller 控制器故障转移（Failover）

Controller 控制器内部设计原理

本质就是一个只能追加写消息的提交日志

Replication 备份

ISR(In Sync Replicas)

Unclean 领导者选举

Replication 备份的作用

ISR

AR

什么情况下一个 broker 会从 ISR 中踢出去 ？

消费者群组的偏移量会写入到`__consumer_offsets topic`的某个分区，该分区的`leader`所在`broker`就是该消费者组的群组协调器

群组协调器的主要作用是

是什么？

Consumer Offset

Kafka如何搭建及创建topic、发送消息、消费消息？

偏移量

__consumer_offsets

提交位移

高水位 High Watermark

是什么 ?

触发条件、触发时机

问题

Coordinator

如何避免

过程

心跳分支来通知

最多一次（at most once）

消息不会被重复发送，最多被传输一次，但也有可能一次不传输。

消息可能会丢失，但绝不会被重复发送

可能丢失 不会重复

至少一次（at least once）

消息不会丢失，但有可能被重复发送。

消息不会被漏发送，最少被传输一次，但也有可能被重复传输.

可能重复 不会丢失

精确一次（exactly once）

消息不会丢失，也不会被重复发送。

不丢失 不重复，只发一次

不会漏传输也不会重复传输,每个消息都传输被一次而且仅仅被传输一次，

这是大家所期望的。

丢消息

kafka收到消息后会写入磁盘缓冲区里，没有直接落到物理磁盘上去，所以机器本身故障了，可能导致磁盘缓冲区里的数据丢失。

主要支持发送消息，消费消息。

其他场景比较受限制。

拓展知识点

（queuing）

一组消费者可以从服务器读取数据记录，每个记录都会被其中的一个消费者处理

 consumers可以同时从服务端读取消息，每个消息只被其中一个consumer读到。

队列模式中，consumers可以同时从服务端读取消息，每个消息只被其中一个consumer读到。

消息被广播到所有的consumer中。

记录被广播到所有的消费者

(publish-subscribe)

发布-订阅模式中消息被广播到所有的consumer中。

offset

MessageSize

data

partition物理上由多个segment文件组成，每个segment大小相等，顺序读写。

每个segment数据文件以该段中最小的offset命名，文件扩展名为.log。

这样在查找指定offset的Message的时候，用二分查找就可以定位到该Message在哪个segment数据文件中。

Kafka为每个分段后的数据文件建立了索引文件，文件名与数据文件的名字是一样的，只是文件扩展名为.index。

index文件中并没有为数据文件中的每条Message建立索引，而是采用了稀疏存储的方式，每隔一定字节的数据建立一条索引。

这样避免了索引文件占用过多的空间，从而可以将索引文件保留在内存中。

图解

每个分区在Kafka集群的若干服务中，都有副本

这样这些持有副本的服务，可以共同处理数据和请求

副本使Kafka具备了容错能力。

副本数量是可以配置的。

leader，负责处理消息的读和写

followers，则去复制leader

如果leader down了，followers中的一台则会自动成为leader。

作为它所持有的一部分分区的leader

作为其他分区的followers

支持在线水平扩展。

因为Kafka内在就是分布式的，一个Kafka集群通常包括多个代理。

为了均衡负载，将Topic分成多个分区，每个代理存储一或多个分区。多个生产者和消费者能够同时生产和获取消息。

消息由生产者通过Kafka Producer API push到Kafka对应Topic中的Partition

再被所关注此Topic的消费者通过Consumer API pull到本地消费

同时Consumer自己管控消息的偏移量offset，即记录此次消息的消息位置。

Kafka的工作原理快速理解

既不是完全的同步复制

也不是单纯的异步复制

要求All Alive Follower都复制完，

这条消息才会被认为commit，

这种复制方式极大的影响了吞吐率

Follower异步的从Leader复制数据，

数据只要被Leader写入log就被认为已经commit，

这种情况下，如果leader挂掉，会丢失数据，

kafka使用ISR的方式很好的均衡了确保数据不丢失以及吞吐率。

点击下载最新的版本并解压.

> tar -xzf kafka_2.9.2-0.8.1.1.tgz

> cd kafka_2.9.2-0.8.1.1

Kafka用到了ZK，所有首先启动ZK

现在启动Kafka

手动创建一个叫做“test”的topic

通过list命令，查看创建的topic

Kafka 使用一个简单的命令行producer，从文件中或者从标准输入中读取消息，并发送到服务端。

默认的每条命令将发送一条消息。

运行producer，并在控制台中输一些消息，这些消息将被发送到服务端：

Kafka也有一个命令行consumer，可以读取消息并输出到标准输出：

> bin/kafka-console-consumer.sh --zookeeper localhost:2181 --topic test --from-beginning

在一个终端中运行consumer命令行，另一个终端中运行producer命令行，就可以在一个终端输入消息，另一个终端读取消息。

这两个命令都有自己的可选参数，可以在运行时，不加任何参数，查看帮助信息。

刚才只是启动了单个broker

现在启动有3个broker组成的集群，这些broker节点也都是在本机上的

（1）为每个节点编写配置文件

（2）在拷贝出的新文件中，添加以下参数

（3）刚才已经启动了ZK和一个节点，现在启动另外两个节点

（4）创建一个拥有3个副本的topic:

（5）现在搭建了一个集群，怎么知道每个节点的信息呢？

（6）向topic发送消息：

（7）消费这些消息：

（8）测试一下容错能力

读写分离不能提高读性能

方便实现“Read-your-writes”

方便实现单调读（Monotonic Reads）

<dependency>
  <groupId>org.springframework.kafka</groupId>
  <artifactId>spring-kafka</artifactId>
</dependency>

KafkaTemplate 主要实现kafka的什么功能？

#============== kafka ===================
# 指定kafka 代理地址，可以多个
spring.kafka.bootstrap-servers=127.0.0.1:9092
#=============== provider  =======================
spring.kafka.producer.retries=0
# 每次批量发送消息的数量
spring.kafka.producer.batch-size=16384
spring.kafka.producer.buffer-memory=33554432
# 指定消息key和消息体的编解码方式
spring.kafka.producer.key-serializer=org.apache.kafka.common.serialization.StringSerializer
spring.kafka.producer.value-serializer=org.apache.kafka.common.serialization.StringSerializer
  

/**
 * Copyright 2018 Desaysv Inc.
 */
package com.desaysv.rd.analysis.utils;

import java.util.List;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Component;

import com.desaysv.rd.analysis.entity.KafkaMsgBean;

import cn.hutool.core.collection.CollUtil;
import cn.hutool.core.util.StrUtil;

/**
 * @Description： 功能描述
 * 
 * @author Xiquan.Liu@desay-svautomotive.com on [2019年5月16日下午7:57:35]
 * @Modified By： [修改人] on [修改日期] for [修改说明]
 *
 */
@Component
public class KafkaUtils {

    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;

    /**
     * @Description： 发送消息到kafka
     * 
     * @author Xiquan.Liu@desay-svautomotive.com
     * @Date 2019年5月17日上午8:57:23
     * @param topic kafka的主题
     * @param json 发送的内容
     */
    public boolean sendMsg(String topic, String json) {
        if (StrUtil.isBlank(topic) || StrUtil.isBlank(json)) {
            return false;
        }
        kafkaTemplate.send(topic, json);
        return true;
    }

    /**
     * @Description： 发送消息到kafka
     * 
     * @author Xiquan.Liu@desay-svautomotive.com
     * @Date 2019年5月30日上午9:07:36
     * @param topic kafka主题
     * @param key 主键
     * @param json 发送的内容
     * @return
     */
    public boolean sendMsg(String topic, String key, String json) {
        if (StrUtil.isBlank(topic) || StrUtil.isBlank(json)) {
            return false;
        }
        kafkaTemplate.send(topic, key, json);
        return true;
    }

    /**
     * @Description： 批量存储
     * 
     * @author Xiquan.Liu@desay-svautomotive.com
     * @Date 2019年6月10日下午5:10:32
     * @param topic 存储的主题
     * @param msgs
     * @return
     */
    public boolean sendBatchMsg(String topic, List<KafkaMsgBean> msgs) {
        if (CollUtil.isEmpty(msgs)) {
            return false;
        }
        msgs.forEach(msg -> {
            sendMsg(topic, msg.getDeviceId(), msg.getMsgContent());
        });
        return true;
    }
}

消费者
标识需要监听主题

Spring的生命周期   

Spring bean初使化与销毁说明

KafkaListener监听流程

KafkaListener使用流程

KafkaListener具体运用示例

高吞吐

低延迟

操作系统层

JVM

Broker

应用层

跨数据中心的传输： 

增加线程

提高 batch.size

增加更多 producer 实例

增加 partition 数

设置 acks=-1 时，如果延迟增大：可以增大 num.replica.fetchers（follower 同步数据的线程数）来调解；

增加 socket 缓冲区设置以及 OS tcp 缓冲区设置。

图解RocketMQ

组（Group）

消息消费者/收件人（Consumer）

消息生产者/寄件人（Producer）

Broker(邮递员)

什么是Namesrv ？概述

Namesrv架构

作用

各个Namesrv节点是无状态的

即每两个Namesrv节点之间不通信，互相不知道彼此的存在。

Namesrv核心数据结构和API

不使用ZK，而是NameServer

NameServer

在Broker、生产者、消费者启动时，轮询全部配置的 Namesrv 节点，拉取路由信息。

支持集群化部署

Peer-to-peer

Broker会注册到NameServer，Producer和Consumer用NameServer来发现Broker

每隔10秒检查Broker的最新心跳时间，如果超过120S都没有发送心跳，则从路由中移除

Producer和Consumer每隔30秒拉取NameServer上的信息。ScheduleAtFixRate

Name Server是RocketMQ的寻址服务。

用于把Broker的路由信息做聚合。用户端依靠Name Server决定去获取对应topic的路由信息，从而决定对哪些Broker做连接。

Name Server是一个几乎无状态的结点，Name Server之间采取share-nothing的设计，互不通信。

Name Server所有状态都从Broker上报而来，本身不存储任何状态，所有数据均在内存。

5.3 RocketMQ的路由原理

注册

心跳机制

主题/地区（Topic）

标签（Tag）

Offset

Message

场景

解决方案

阿里出品

Apache

开源

Java语言开发

功能完备

顺序消息

顺序消息

Hash取模

Broker非正常关闭
Broker异常Crash
OS Crash
机器掉电，但是能立即恢复供电情况
机器无法开机（可能是cpu、主板、内存等关键设备损坏）
磁盘设备损坏

消息回溯

可查询

发送成功后返回consume_success

回溯消费

消费保证

可以保证高可用性

高可用性

主从同步

故障转移

高可用

支持集群模式

集群化部署

可以部署大规模的集群。

性能很高

吞吐量很高，单机可达10万QPS以上

亿级堆积

支持通过配置保证数据不丢失

丰富的消息机制

Rocket MQ提供给我们push|pull 两种消息拉取机制

长轮询机制

master根据自己的负载，向consume发送建议，是否从slave拉取消息

slave同步落后过多时，master建议consume只从master拉取消息

slave挂掉对整体有点影响不过不大

master挂掉需要运维工程师手动调整配置，把slave切换成master，不支持自动主备切换

master-slave不是彻底的高可用，无法实现自动主备切换，version 4.5后，引入了Dledger，实现了高可用自动切换

produce

broker

consume

零拷贝

文件系统

数据存储结构

降低系统耦合

提高接收能力

异步化提升性能

系统A发送过来的每秒1万请求是一个流量洪峰，然后MQ直接扛下来了，都存储在本地磁盘，

这个过程就是流量削峰的过程，瞬间把一个洪峰给削下来了，让系统B后续慢慢获取消息来处理。

1.1 消息存储整体架构

1.2 页缓存与内存映射

1.3 消息刷盘

2.1 Remoting通信类结构

2.2 协议设计与编解码

2.3 消息的通信方式和流程

2.4 Reactor多线程设计

4.1 Producer的负载均衡

4.2 Consumer的负载均衡

5.1 RocketMQ事务消息流程概要

5.2 RocketMQ事务消息设计

6.1 按照MessageId查询消息

6.2 按照Message Key查询消息

目录解析

在源码中打一些端点，观察rocketmq源码运行过程，然后在这个过程中，需要从rocketmq实际运行和使用的角度，去观察他的源码运行流程。概括来讲可以分为三点：场景驱动+通俗语言+大量画图。

场景驱动

通俗语言

大量画图

启动类

关键类

配置类

结合NameServer启动日志观察其启动过程。

总结

启动类

关键类

配置类

总结

Dledger快速搭建(Quick Start) :

DledgerCommitLog代替Broker管理CommitLog

基于Raft协议选举Leader Broker

基于Raft协议进行多副本同步

Leader Broker崩溃的话怎么办

防止程序bug等导致疯狂写入MQ消息的情况

可采用令牌桶算法等来进行限制

双写+双读方案同步数据

一个是从Topic的第一条数据开始消费。

一个是从最后一次消费过的消息之后开始消费。

权限管理(Auth Management)

配置

通过在broker端放一个acl权限文件，在里面规定好权限，哪个消费者对应哪个topic有什么操作权限等

生产环境消息轨迹追踪

消息轨迹指南(Message Trace):

broker

producer/consumer

broker的配置文件里开启traceTopicEnable=true

通过rocket-console上面的消息轨迹，创建查询任务，也可以看到消息的轨迹

CDN+Nginx，Lua脚本+Redis三级缓存处理静态页面

下单前加入答题模块，避免脚本疯狂调用接口以及错峰

独立部署一套秒杀响应的服务

redis控制扣减库存，快速响应

抢购完毕时，Lua脚本过滤无效请求

瞬时流量进入rocketmq削峰

难点分析

如何解决秒杀活动压力过大的问题

可以设置consumer端的参数：consumeThreadMin，consumeThreadMax，这样一台consumer机器上的消费线程越多，消费的速度就越快。

可以开启消费者批量消费的功能，设置consumeMessageBatchMaxSize参数，默认是1，设置的多一些，就会交给你的回调函数一批消息处理

一般如果消息不重要的话就在consume上直接释放掉

如果topic的messageQueue设置的比较多，比如设置了20个，consume实例只有4个，那么每个consume实例对应5个messageQueue，这个时候可以申请临时加机器，增加consume实例为20个，达到快速消费的目的

如果messageQueue设置的比较少，比如只设置了4个，那么这个时候就不能通过加consume机器来解决了，这时候就需要修改消费者代码了，不再消费者消费，而是把要消费的消息放到mq的另一个topic中，这个topic设置20个messageQueue，对应20个consume实例，进行消费

介绍

开源技术方案

数据同步

具体方案

如果消息允许丢失，就可以紧急修改消费者系统代码，在代码中对所有消息都获取到的就直接丢弃，不做任何处理，这样可以迅速的让积压在MQ的百万消息被处理掉，只不过处理方式是全部丢弃而已。

临时申请16台机器多部署16个消费者系统的实例，然后20个消费者同时消费，每个人消费一个MessageQueue的消息，此时你会发现消费的速度提高五倍，很快积压的百万消息都会被处理完毕。处理完成后，直接下线新加的机器。

如果messageQueue数量有限，只有4个，然后就没办法扩容消费者系统，因为加再多的消费者系统，还是只有4个messageQueue，没法并行消费，所以可以临时修改4个消费者的代码，让他们获取消息后不写nosql，而是直接把消息写入新的topic，这个速度很快，因为仅仅只是读写mq,然后新的messageQueue有20个，可以再部署20台临时追加的消费者系统，去消费新的topic后写入数据到NoSQL里去，这样子也可以迅速增加消费者的并行处理能力，使得用一个新的topic来允许更多的消费者系统并行处理。

try-catch机制，发送消息到MQ，失败的话捕获，重试3次，依旧失败则认为MQ集群崩溃

这个时候需要把消息有序的放到DB/NO SQL中，待MQ集群恢复正常后，再进行消费

注意：此方案保证消息写入有序

通常会在你发送消息到MQ的那个系统中设计高可用的降级方案，这个降级方案通常的思路是，需要在你发送消息到MQ代码里去try catch捕获异常，如果你发送消息到MQ有异常，此时你需要进行重试。

如果连续重试3次还是失败，说明此时可能就是你的MQ集群崩溃了，此时你必须把这条重要的消息写入本地存储中去，可以说是写入数据库中，也可以是写入到机器本地磁盘文件里，或者noSQL存储里，几种方式我们都做过。之后只要你不停的发送消息到MQ去，一旦发现MQ集群恢复了，你必须有一个后台线程可以把之前持久化存储的消息都查询出来，然后依次按照顺序发送到MQ集群里去，这样才能保证你的消息不会因为MQ奔溃而彻底丢失。

你把消息写入存储中暂存时，一定要保证他的顺序，比如按照顺序一条一条的写入本地磁盘文件去暂存消息。一旦集群故障了，你后续的所有写消息代码必须严格按照顺序把消息写入到本地磁盘文件里暂存，这个顺序是要严格保证的。只要有这个方案在，哪怕你的MQ集群突然崩溃了，你的系统也是不会丢失消息的。

接收消息时，必须引入限流机制，可以采用令牌桶算法。

如果你要做MQ集群迁移，是不可能简单粗暴的，在某个时间点突然之间就说把所有的producer系统都停机了，然后更新他的代码，接着全部重新上线，然后所有的Producer系统都把消息写入到RocketMQ里去了。

一般来说，首先你要做到双写，也就是说你所有的Producer系统中，要引入一个双写的代码，让他同时往kafka和rocketmq中写入消息，然后多写几天，起码双写要持续1周左右，因为mq一般都是实时数据，里面数据最多保留一周。

当你双写持续一周过后，你会发现你的kafka和rocketmq里面的数据看起来几乎一模一样了，因为MQ也只保留近几天的数据。

但是光双写还是不够的，还需要同时进行双读，也就是说在你双写的同时，你所有的consumer系统都需要同时从kafka和rocketmq获取消息，分别用一摸一样的逻辑处理一遍，只不过从kafka获取到的消息还是走核心逻辑去处理，然后落库或者别的存储什么的，但是对于rocketmq用一样的逻辑处理，只是不落库。

你的consumer系统在同时从kafka和rocketmq进行消息读取的时候，需要统计每个mq当日读取和处理的消息数量，这点非常重要，同时对于rocketmq读取的消息处理之后的结果。可以写入临时存储中。

当你观察一段时间，发现双写和双读一段时间之后，如果所有的consumer系统通过对比发现，从kafka和rocketmq读取和处理的消息数量一致，同时处理之后得到的结果一致，此时就可以判断说当前kafka和rocketmq里的消息一致，可以计算出来的结果也是一致的。

这个时候就可以实施正式的切换了，可以停机Producer，再重新修改上线后，全部修改为仅仅写rocketmq这个时候数据也不会丢失，因为之前已经双写了一段时间后，然后所有的consumer系统可以全部下线后，修改代码再上限，全部基于rocketmq来获取消息，计算和处理，结果写入存储中。

设置tag，consume进行消费的时候可以根据tag进行过滤

RocketMQ还是支持比较丰富的数据过滤语法的

mqadmin queryMsgByKey -n 127.0.0.1:9876 -t SCANERCORD -k orderId

消息发送零丢失

消息消费零丢失

总结

生产配置以及压测

5 客户端配置

6 系统配置

9.1 RocketMQ落地概述

9.2 RocketMQ集群管理

9.3 RocketMQ集群监控和报警

9.4 RocketMQ集群迁移

9.5 RocketMQ测试环境实践

9.6 RocketMQ接入实践

集群部署(Cluster Deployment):

4 种高可用 RocketMQ 集群搭建方案！

集群搭建

 集群搭建

mqadmin管理工具

运维常见问题

3.1消费者概述

3.2消费者启动机制

3.3消费者的Rebalance机制

3.4消费进度保存机制

3.5消费方式

3.6消息过滤

3.7消费者最佳实践总结

2.1生产者原理

2.2生产者启动流程

2.3消息发送流程

2.4发送消息最佳实践

2.5生产者最佳实践总结

接下来我们先在linux平台下安装一个RocketMQ的服务

环境准备

安装RocketMQ

启动RocketMQ

测试RocketMQ

关闭RocketMQ

物理部署

逻辑部署

4.1 RocketMQ架构

4.2常用的部署拓扑和部署实践

下载

修改配置文件

打成jar包，并启动

访问控制台

1) 基本样例

2) 顺序消息样例

3） 延时消息样例

)4批量消息样例

5过滤消息样例

6消息事务样例

7 Logappender样例

8 OpenMessaging样例

Advanced Message Queue

高级消息队列协议

一个异步消息传递所使用应用层协议规范

为面向消息中间件设计

基于此协议的客户端与消息中间件可传递消息，并不受产品、开发语言等条件的限制。

基于此协议的客户端与消息中间件可以无视消息来源传递消息，不受客户端、消息中间件、不同的开发语言环境等条件的限制；

AMQP 的出现其实也是应了广大人民群众的需求

RabbitMQ遵循 AMQP 协议

使用Erlang编写的

使用erlang语言开发

一个由 Erlang 语言开发的 AMQP 的开源实现

一个由erlang开发的AMQP的开源实现

RabbitMQ是采用erlang语言开发的，所以必须有erlang环境才可以运行

RabbitMQ 最初起源于金融系统

RabbitMQ是由RabbitMQ Technologies Ltd开发并且提供商业支持的。

该公司在2010年4月被SpringSource收购。

在2013年5月被并入Pivotal。

其实VMWare，Pivotal和EMC本质上是一家的。

不同的是VMWare是独立上市子公司，而Pivotal是整合了EMC的某些资源，现在并没有上市。

更适合企业级的开发

本身支持很多的协议：AMQP，XMPP, SMTP, STOMP

开源

一种消息中间件

用于在分布式系统中存储转发消息，在易用性、扩展性、高可用性等方面表现不俗。

对路由，负载均衡或者数据持久化都有很好的支持

同时实现了Broker构架，这意味着消息在发送给客户端时，先在中心队列排队。

性能较好

适合于企业级的开发

消息的生产者，也是一个向交换器发布消息的客户端应用程序。

表示消息队列服务器实体。

接收客户端连接，实现AMQP协议的消息队列和路由功能的进程

Broker（代理/中介）

就是rabbitmq服务，默认端口5672

虚拟主机，表示一批交换器、消息队列和相关对象。虚拟主机是共享相同的身份认证和加密环境的独立服务器域。

虚拟主机的概念，类似权限控制组

一个Virtual Host里可以有多个Exchange和Queue

Vhost（Exchange+Message Queue）

Exchange（将消息路由给队列 ）

交换器，用来接收生产者发送的消息并将这些消息路由给服务器中的队列。

交换机，接收生产者发送的消息，并根据Routing Key将消息路由到服务器中的队列Queue。

交换机，通过绑定键（Binding Key）与Queue绑定

ExchangeType、Exchange 类型、路由方式

（消息队列和交换器之间的关联）

绑定，用于消息队列和交换器之间的关联。

一个绑定就是基于路由键将交换器和消息队列连接起来的路由规则，所以可以将交换器理解成一个由绑定构成的路由表。

BindingKey

Queue

队列，存储消息

消息队列，用来保存消息直到发送给消费者。

它是消息的容器，也是消息的终点。

一个消息可投入一个或多个队列。

消息一直在队列里面，等待消费者连接到这个队列将其取走。

消息队列，用于存储还未被消费者消费的消息；

消息，消息是不具名的，它由两部分组成

网络连接，比如一个TCP连接。

生产者和消费者与Broker建立的TCP长连接

信道，多路复用连接中的一条独立的双向数据流通道。

信道是建立在真实的TCP连接内地虚拟连接，AMQP 命令都是通过信道发出去的，不管是发布消息、订阅队列还是接收消息，这些动作都是通过信道完成。

因为对于操作系统来说建立和销毁 TCP 都是非常昂贵的开销，所以引入了信道的概念，以复用一条 TCP 连接。

虚拟连接，减少TCP创建和销毁的性能损耗，api编程最主要接口

消息的消费者，表示一个从消息队列中取得消息的客户端应用程序。

高可靠

Reliability

消息不丢失

可以保证数据不丢失。

RabbitMQ 使用一些机制来保证可靠性

可靠性投递

灵活的路由

Flexible Routing

在消息进入队列之前，通过 Exchange 来路由消息的。

对于典型的路由功能，RabbitMQ 已经提供了一些内置的 Exchange 来实现。

针对更复杂的路由功能，可以将多个 Exchange 绑定在一起，也通过插件机制实现自己的 Exchange 。

集群

高可用

Multi-protocol

RabbitMQ 支持多种消息队列协议

本身支持很多的协议：AMQP，XMPP, SMTP, STOMP、MQTT

多客户端支持

Many Clients

与Spring集成

RabbitMQ 几乎支持所有常用语言，比如 Java、.NET、Ruby 等等

可以跨平台、跨语言使用。

权限管理

可视化管理界面

简洁易用的可视化管理界面

Management UI

RabbitMQ 提供了一个易用的用户界面，使得用户可以监控和管理消息 Broker 的许多方面。

图解RabbitMQ 可视化管理界面

Tracing

如果消息异常，RabbitMQ 提供了消息跟踪机制，使用者可以找出发生了什么。

Plugin System

插件系统

RabbitMQ 提供了许多内置插件，来从多方面进行扩展

RabbitMQ也可以编写自己的插件

死信队列

延迟队列

消息过期时间TTL

死信队列

消息重试

消费端限流

服务端流控

示例代码

示例代码

//开启手动应答
channel.basicConsume(QUEUE_NAME, false, defaultConsumer);

//执行手动应答
channel.basicAck(envelope.getDeliveryTag(), false);

下载并安装erlang：http://www.erlang.org/download
配置erlang环境变量信息

新增环境变量ERLANG_HOME=erlang的安装地址
将%ERLANG_HOME%\bin加入到path中

下载并安装RabbitMQ，下载地址：http://www.rabbitmq.com/download.html
注意: RabbitMQ 它依赖于Erlang,需要先安装Erlang。

docker安装

docker pull rabbitmq:3-management

docker run -d --hostname my-rabbit --name rabbit -p 15672:15672 -p 5672:5672 rabbitmq:3-management

docker run -d --hostname my-rabbit --name rabbit -e RABBITMQ_DEFAULT_USER=user -e RABBITMQ_DEFAULT_PASS=password -p 15672:15672 -p 5672:5672 rabbitmq:3-management

https://www.cnblogs.com/yufeng218/p/9452621.html

单机安装

java原生api操作

UI管理界面

Spring AMQP介绍

Spring AMQP核心组件

Spring集成RabbitMQ配置文件解读

Spring Boot集成RabbitMQ

SpringBoot整合RabbitMQ实现消息队列技术1

SpringBoot整合RabbitMQ实现消息队列技术2

Storm是什么？

Storm的底层

Saltstack由Master和Minion构成，通过ZeroMQ进行通信

使用消息队列zeroMQ传输数据

入队

出队

软实时 (Soft-Realtime)

低延时 (Low-Latency)

分布式 (Distributed)

子主题

在这种结构下，如果订阅节点发⽣故障，就会导致发布者的消息丢失（QoS 0）或者堆积在 Server 中（QoS 1, 2）。

⼀般情况下，解决这个问题的办法都是直接增加订阅节点，但这样⼜产⽣了⼤量的重复消息，

不仅浪费性能，在某些业务场景下，订阅节点还需要⾃⾏去重，进⼀步增加了业务的复杂度。

⽬前两种实现⽅式

利⽤中间件将有状态服务转化为⽆状态服务

不同的消息可以路由⾄不同的订阅⽅，能够实现消息分⽚，

并且如果当某⼀个订阅⽅出现宕机，对应的消息会分配给存活的订阅⽅，从⽽能够实现⾼可⽤。

共享策略