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、不同的消息队列收到的消息可能是无序的

优点

缺点

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

优点

缺点

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

消息不会重投

降低拉取频率

生产者生产速度过快；

短时间的业务高峰期；

Broker同步策略导致消费堆积

消费者消费速度过慢

消费者宕机

生产者限流

调整刷盘策略

多线程消费消息

设置数据开关，开关打开消息直接放入数据库，或者直接返回，最大程度上降低消费者程序时间

的差别值与给定的消息堆积数值告警值对比，若是差别   值大于数据告警值，则存在消息堆积，不然不存在消息堆积

producer消息的发送速度大于consumer的消息消费速度

producer的生产速率无明显增长，consumer的消费速率无明显增长

producer生产速率正常，RocketMQ服务器性能正常，consumer消费速率下降

最佳工作模式：RocketMQ自己的服务性能，必要的时候能够对RocketMQ 进行扩容，提升消息堆积能力。

不需要考虑消费顺序

消费者和消费者同时都只有一台机器

生产者消费消息的时候，把消息发送到同一个topic下同一个队列；可以保证消费方只有一个线程去消费消息

将需要消费的顺序消息，合并成一条消息发送出去，这样消费的时候就是有顺序；

设置发送的消息为全局顺序消息

1、消费者没有发出

2、网络原因导致数据丢失

3、rocketMQ节点没有收到

消费者代码逻辑中保持幂等性

通过每条消息的唯一编号来保证

有些业务场景，重复接受到消息，也不会影响到业务，所以不处理也行

1、生产者发送到队列节点消息丢失

2、RocketMQ节点消息未能持久化到磁盘

3、RoekctMq节点消费者丢失

RoekctMQ  事物机制来确保消息都能发送到节点上

消息重投机制，消息投递失败，会再次投递

1、修改节点的消息持久化刷盘机制

2、RocketMQ采用主从机构

ack机制，设置消息成功消费之后，再通知节点消息已经成功消费

性能和吞吐量也将大幅下降

使用事务机制传输消息

同步刷盘

主从机制

消费完再通知节点