MQ

Headers Exchange默认交换机

Fanout Exchange广播交换机

Direct Exchange直连交换机

Topic Exchange主题交换机

name

Durability

Auto-delete

Arguments（拓展参数）

表示作用

处理方式

发送端把消息发送到服务器，结果找不到对应的交换机，路由队列；return消息机制就是应对这种情况

发送消息时候给Channel参数manDetory设置为true；消费就会返回到发送端可以做后续处理；如果为false,服务端就会直接把该条消息删除

实现原理：AMQP协议

发送端把消息发送个服务端，服务端接收到消息并且把消息持持久化到磁盘就会给发送端一个异步的confirm应答

confirm种类

ack种类模式

开启重试机制

最大重试次数

重试间隔时间

最大间隔时间（不能超过这个时间间隔）

间隔时间乘法数（重试的时间间隔在上一次的倍数）

把交换机的属性持久化；在宕机或者重启之后服务器可以自动的去创建交换机，避免手动或者跑程序创建

设置durable=true

把队列的属性持久化，在宕机或者重启之后可以自动的去创建队列，避免手动创建

设置durable=true

消息的持久化是建立在队列的持久化之上，如果队列没有持久化，那么消息也不能持久化

设置 deliveryMode =2 ； deliveryMode =1 是不进行持久化

所有队列的消息都会写入到磁盘的中间中去；当写入的数据大小超过了文件大小，那么就会关闭此文件，再新建一个文件存储；

消息本身推送到消费端的时候在服务端需要存入磁盘

内存资源少，需要把队列中的数据存入磁盘

消息并不是来一条消息就往磁盘上存储一条，而是先把消息都放入到一个缓冲池；等一定的条件才会缓存的消息写入磁盘

1.缓冲池缓冲的数据大小超过缓冲池本身

2.超过固定的刷盘时间25ms，不管缓冲池是否满了，都会刷盘

3.消息写入缓冲区后，没有其他后续请求写入，那么也会刷盘

收到消费者的ack消息的时候，并不是马上去删除消息，而是先给消息做一个删除的标记

后台进程检车到垃圾数据比例超过50%，并且文件不少于3个，的时候就会触发持久化数据的垃圾回收；找到符合要求的左右两个文件，先整理左文件中的有效数据，然后再把有文件中有消息数据复制到左文件;再把又文件删除；

1.所有文件中垃圾数据达到50%的比例；

2.存储的文件必须至少有三个；

消费在队列中存活时间；当时间超过了消息就会被抛弃；设置死信交换机，被抛弃的消息就会落入到死信交换机；

1.不设置消费者，就可以让消息一直堆积，直到超过存活时间

1.创建死信交换机

2.创建死信路由

3.新建消费者队列绑定死信路由

1.死信交换机就是普通交换机

2.死信交换机被动接受其他交换机或者无法消费的消息

3.创建生产的交换机的时候就需要设置对应的死信交换机

丢失原因

解决方案

丢失原因

解决方案

丢失原因

解决方案

生产者突然发送大量信息

消费者消费失败

消费者出现性能瓶颈

消费者直接挂掉

设置多个线程同时处理消费消息

设置一次从服务端拉取多条消息

生产者拆分成多个，让生产者和消费者一对一生产消费（消费者内部可以开多线程消费）

开启多个消费者，把前后有关联的数据往同一个消费者发送

1.全局消息id做幂等性判断

2.全局业务id做幂等性判断

1.一主一备；也可以是一主多备

2.主节点提供读写，从节点备份主节点数据

3.主节点挂了，从节点就会变成主节点；原来的从节点回复之后，就会变成备用节点

1.并发和数据量不高的情况下；

1.需要使用haproxy作为中间件

DMA直接内存访问；现代计算机就是允许硬件之间直接进行数据交互；DMA将一个地址空间复制到另外一个地址空间，然后数据的传输是DMA设置之间完成

DMA

DMA设备

mmap

sendfile

Kafka的零拷贝并不是说完全不存在拷贝，而是避免不必要的拷贝

从磁盘把数据拷贝到内核空间；

从内核空间中直接把数据发送到网卡；

1.从磁盘去读到内核空间缓存页；

2.应用从内核缓存页读取到用户空间缓存区；

3.应用程序将用户缓冲区的数据放入socket缓冲区；

4.操作系统将socket里面的数据复制到网卡接口，发送数据；

1.kafka的零拷贝从获取数据到最终把数据发送出去只需要经历一次拷贝；

2.传统拷贝方式从获取数据到最终把数据发送出去，需要经历4次拷贝；

kafka的拷贝方式大大降低了数据在不同的内存空间中复制的次数，提高了系统io效率

基于磁盘的线性的读写（操作系统做了大量的IO技术优化），甚至会被随机的内存读写更快

io优化技术

kafka是直接把数据写入日志文件；其他几乎都是先把数据缓存在内存中然后再间隔刷盘

读写操作

读写概述

好处

删除策略

删除内容

删除操作阻塞读操作

日志文件特点

日志条目

日志文件名称

记录每一个segment下包含的日志条目偏移量范围

在消息有效期内，是允许消费者重复消费；

日志删除

日志压缩

1.消息可以指定分区发送

2.消息可以通过负载均衡方式发送到不同的分区

3.通过指定key进行hash运算后确定让哪个分区发送

消费者集群的各个消费者只能消费不同的分区

一个topic消息可以发送给多个消费者集群

一个消费者可以消费多个集群的消息

多个消费者集群可以消费一个topic下面的消息

注册服务节点

注册topic

注册消费者

生产者负载均衡

消费者负载均衡

记录分区与消费者关系

记录消费中的偏移量

此负载均衡是kafka自带的

缺点是无法动态感知服务器节点的新增减少，从而在服务器新增减少的时候，不能根据服务器做负载均衡

有新的分区上线就重新选leader

运行重新分区命令，重新选择leader

运行重新选择leader命令

服务正常关闭之后，重启重新选择leader

leader副本的数据和其他副本数据都不一致，读写分离容易导致数据不一致

leader副本数据到从副本数据有数据延迟；

主要原因就是消息消费时候提交的偏移量，服务器并不知情

1.强行杀掉线程，导致偏移量没有提交

2.消费了，还没有提交偏移量，分区就掉线了，触发重平衡，然后消息就会重复消费

3.消费者重新分配分区，导致消费者数据重新消费

4.消息消费时间过长，让zk觉得机器宕机了，触发了重平衡

副本同步leader数据

同步参数

可以设置一个topic只有一个分区，只有一个消费者

生产者把需要顺序消费的消息发送到指定的分区上

消息丢失原因

消息丢失原因

消息丢失原因

消息确认机制

消息重试机制

消息确认机制和重试机制区别

1.ACK=all，让所有的服务器节点都获取到数据；

2.合理的设置从机的个数，设置数据的备份

3.禁止主机挂掉，选从机作为新的主机

设置消息消费提交偏移量为手动提交偏移量，通过代码在finnaly里面手动设置消息异常的那个前一条的偏移量做提交；

添加分区个数和消费者个数

合理增大每次拉取的消息数量

分区数量- 3，消费者数量- 3

Kafka 将一个分区分配给一个使用者。除非某些使用者发生故障并且发生使用者重新平衡（将分区重新分配给使用者），否则所有使用者都将映射到其分区，并按顺序使用这些分区的事件。

分区数量 - 1，使用者 - 3

如果消费者多于分区数量，Kafka就没有足够的分区来分配消费者。因此，该组中的一个消费者被分配给分区，而该组中的其他消费者将处于闲置状态。

分区- 4，消费者- 3

在此方案中，其中一个使用者获得 2 个分区，而在使用者重新平衡期间，另一个使用者可能会获得 2 个分区。

broker在集群中的标识

默认值-1

监听的服务地址(多个用,隔开)

无默认值

连接的zookeeper地址(多个地址用,隔开)

连接zookeeper超时时间(毫秒)

无默认超时时间

连接ZK会话超时时间

zk的从机落后zk主机的最长时间

消费者有多少个未确认的消息，才会导致阻塞

日志存放目录(有多个目录分布时使用,隔开)

无默认值

日志存放目录(当log.dirs为null时)

默认值/tmp/kafka-logs

将消息刷新到磁盘之前，日志分区上累计的消息数量

默认值：9223372036854775807

刷盘前在内存中最长存在时间

日志文件的最大容量

默认值-1，也就是可以无穷大

log.retention.hours

log.retention.minutes

log.retention.ms

log.roll.hours

log.roll.ms

log.segment.bytes

log.segment.delete.delay.ms

拉取的批量消息的最大内存大小

默认值：0.9M

第一次发动消息时，自动创建topic。

默认值：true;

是否可以删除topic

默认值：true

如果为Flase，那么管理工具将不能删除主题

分区重平衡检查的频率

触发重平衡比例

默认值100%

后台处理线程个数

默认值10；

处理请求线程数量

默认值：8

处理网络请求网络相应线程数量

默认值3

日志恢复和日志关闭时刷新的线程数

默认值1

日志之间移动副本线程数

无默认值

主节点数据复制到副本的线程数

与偏移量提交管道的元数据最大大小

偏移量超时时间

偏移量提交主题分区的数量

子主题 1

日志索引文件大小

默认值100M

按照给定的压缩方式压缩数据

值集：“gzip”、“snappy”、“lz4”、“zstd”

事务执行最长时间，超时则抛出异常

900000ms

服务器节点配置

消息缓冲区大小

默认值：33554432 =32M

生产者最大可以用缓存；生产者可以用来缓冲等待发送到服务器的记录的总内存字节

key.serializer

value.serializer

消息发送条件

消息发送阻塞

消息请求阻塞时间

消息发送大小

acks

delivery.timeout.ms

retries

retry.backoff.ms

compression.type

connections.max.idle.ms

max.in.flight.requests.per.connection：

metric.reporters

interceptor.classes

receive.buffer.bytes

send.buffer.bytes

事务ID(当有多个生产者时，标识哪个生产者的事务,可用于消息幂等)

事务超时时间

服务器连接地址

消费者组的ID

消费者ID

初始偏移量当前偏移量不存在的时候，消费者消费的起始点

值集

默认值

是否公开topic内部的元数据信息

默认值：true；

isolation.level

max.poll.records

max.poll.interval.ms

fetch.max.bytes

fetch.min.bytes

fetch.max.wait.ms

enable.auto.commit

auto.commit.interval.ms

key.deserializer

value.deserializer

connections.max.idle.ms

request.timeout.ms

session.timeout.ms

heartbeat.interval.ms

最大程度上确保消息的不丢失

使用场景

使用场景

使用场景

默认的消费方式，但是实时性不高，但是不会造成消息消费堆积

优点

缺点

本质

1、服务器路由提供者，

2、生产者，消费者能够通过名称服务查询各主题的相应元数据信息

1、多个Name Serve 组成集群

2、集群中各个Name Server相互独立，没有信息交互

1、消费者通过同一个消息队列（topic分区）收到的消息是有序的

2、不同的消息队列收到的消息可能是无序的

优点

缺点

对程序性能要求高，但是顺序消费要求不高

优点

缺点

对消费有顺序要求，且对程序性能要求不高

消息不会重投

降低拉取频率

Producer原因

Broker消息堆积

Consumer原因

Producer原因

Broker消息堆积

Consumer

Producer发送消息的速率监控

Consumer消费消息的速率监控

Producer发送消息的最大偏移量（maxOffset）与Consumer消费消息的当前偏移量（currOffset）

差别值呈现增大趋势

差别值呈现平稳趋势或者降低趋势

某些特殊场景下，发送出去的消息，消费者需要按照顺序来消费

发送出去的多条消息，都是走的同一个topic发送

大多数业务场景不需要考虑消息的顺序性

具体实现

非顺序消息 → 多线程并发消费（线程池处理）。

顺序消息

网络问题，导致消息消费的确认消息，rocketMQ节点没有收到

1、代码层面

2、消息消费层面

3、重复消息不处理

主要有三种场景

针对场景1处理

针对场景2处理

针对场景3处理

导致问题

优化机制

负责 发送消息 到 Broker。

特点：

作用：从 Broker 消费消息。

模式：

特点：

作用：

特点：

类型：

作用：

特点：

部署方式

作用：消息的分类标签

消费组消费 Topic 中的消息时，实际上是消费它的各个队列

作用：

特点：

队列数量由创建 Topic 时指定（或者通过 Broker 配置的默认值）

类型

没有特殊属性，直接发送、存储、消费。

局部顺序（同一个业务的key的消息进入到同一个队列中）

实现方式

机制：

消息不会马上投递，而是等到指定的时间点后才可被消费。

RocketMQ 内部是通过 定时轮询机制 + 延时等级（level） 来实现的（默认支持 18 个固定延时等级，比如 1s、5s、10s、30s、1m、2m、…、2h）。

应用场景：

用于分布式事务场景。

生产者可以一次发送一批消息，减少网络开销，提高吞吐量

同一个消息只会被组内的一个服务实例消费

多台服务器之间实现 负载均衡，消息被均摊。

同一个消费组的每条消息 每个实例都会消费一次

不做负载均衡，每个实例都收到完整消息

Topic 有 4 个队列：Q0、Q1、Q2、Q3

消费组有 2 个消费者：C1、C2

消费者
分配队列
C1
Q0、Q1
C2
Q2、Q3

CommitLog（消息日志）

ConsumeQueue（消费队列）

IndexFile（索引文件）

生产者发消息到broker节点，节点接受到消息写入commitlog中

写入commitlog的时候是按照顺序写入，然后ConsumeQueue去记录该条消息在commitlog中的物理位置偏移量

如果有设置消息 Key，还会写入 IndexFile。

同步刷盘：消息必须写入磁盘成功才返回 → 高可靠。

异步刷盘：写入 PageCache(内存中) 就返回，后台线程再批量写盘 → 高性能。

拉取的起始位置（偏移量）

是否有新消息

轮询频率

消费者完全掌握主动权，能灵活控制速率。

适合定时批量拉取、低实时性场景。

需要自己写轮询逻辑。

实时性差，可能“拉不到”消息。

使用简单，开发者只需要写回调逻辑。

实时性好，消息几乎一到 Broker 就会被消费。

速率由 Broker 控制，Consumer 可能被“推爆”，需要消费端限流。

先把消息发送broker节点上，存储为半消息，消费者对半消息不可见

生产者端执行本地业务逻辑，事物完成之后给broker发送事物状态是成功了，还是需要回滚

broker长时间得不到生产者发送的事物消息状态，则会回查生产者该事物消息的状态。

负责接收生产者发送的消息，储存消息，提供给消费者消费

同步主机的数据，可用于高可用，负载分担消息

流程

优点：保证消息 不丢失（Master 挂掉，Slave 也有完整数据）。

缺点：性能较低，写入延迟高，因为要等 Slave 完成同步。

流程

缺点

优点：吞吐量高，写入延迟低。

Broker 启动时向 NameServer 注册自己的地址和 Topic 信息。

Producer/Consumer 启动或发送/拉取消息时，从 NameServer 查询 Broker 的路由信息。

从机会复制主机的消息，备份主机的消息，保证消息不丢失

除了顺序消费，普通消息是可以从从机拉取消费，提高系统可用性

主机宕机，从机可以切换成主机，保证服务可用