MQ 常见面试题
基础
什么是消息队列
消息队列(Message Queue,简称MQ),指保存消息的一个容器,本质是个队列。
- 消息(Message)是指在应用之间传送的数据,消息可以非常简单,比如只包含文本字符串,也可以更复杂,可能包含嵌入对象。
- 消息队列(Message Queue)是一种应用间的通信方式,消息发送后可以立即返回,有消息系统来确保信息的可靠专递,消息发布者只管把消息发布到MQ中而不管谁来取,消息使用者只管从MQ中取消息而不管谁发布的,这样发布者和使用者都不用知道对方的存在。
一般包含以下三个角色:
- Producer:消息生产者,负责产生和发送消息到 Broker;
- Broker:消息处理中心。负责消息存储、确认、重试等,一般其中会包含多个 queue;
- Consumer:消息消费者,负责从 Broker 中获取消息,并进行相应处理;
为什么要用MQ?
- 屏蔽异构平台的细节:发送方、接收方系统之间不需要了解双方,只需认识消息。
- 异步:消息堆积能力;发送方接收方不需同时在线,发送方接收方不需同时扩容(削峰)。
- 解耦:防止引入过多的API给系统的稳定性带来风险;调用方使用不当会给被调用方系统造成压力,被调用方处理不当会降低调用方系统的响应能力。
- 复用:一次发送多次消费。
- 可靠:一次保证消息的传递。如果发送消息时接收者不可用,消息队列会保留消息,直到成功地传递它。
- 提供路由:发送者无需与接收者建立连接,双方通过消息队列保证消息能够从发送者路由到接收者,甚至对于本来网络不易互通的两个服务,也可以提供消息路由。
消息队列有什么优点和缺点?
核心优点
- 解耦
- 异步
- 削峰
缺点
- 系统可用性降低:系统引入的外部依赖越多,越容易挂掉。
- 系统复杂度提高了
- 一致性问题:消息传递给多个系统,部分执行成功,部分执行失败,容易导致数据不一致
RabbitMQ
RabbitMQ 和 AMQP 是什么关系?
RabbitMQ 和 AMQP 有着非常密切的关系,但是他们是属于完全不同的两个概念。
- AMQP: AMQP 不是一个具体的消息中间件产品,而是一个协议规范。他是一个开放的消息产地协议,是一种应用层的标准协议,为面向消息的中间件设计。AMQP 提供了一种统一的消息服务,使得不同程序之间可以通过消息队列进行通信。 SpringBoot 框架默认就提供了对 AMQP 协议的支持。
- RabbitMQ:RabbitMQ则是一个开源的消息中间件,是一个具体的软件产品。RabbitMQ 使用 AMQP 协议来实现消息传递的标准,但其实他也支持其他消息传递协议,如 STOMP 和 MQTT。RabbitMQ 基于 AMQP 协议定义的消息格式和交互流程,实现了消息在生产者、交换机、队列之间的传递和处理。
总之,AMQP 本质上是一个开放的标准,他不光可以被 RabbitMQ 实现,也可以被其他产品实现。
通过这种标准的协议,实际上是可以在不同的消息中间件系统之间进行灵活的消息传递。只不过,目前具体实现这种标准的产品目前并不多,RabbitMQ 则是最有影响力的一个产品。因此,RabbitMQ 成了 AMQP 协议事实上的代表。SpringBoot 框架默认提供的 AMQP 协议支持底层也是基于 RabbitMQ 产品实现的。
RabbitMQ 的核心组件有哪些?
RabbitMQ的核心组件包括以下几部分,他们共同构成了 RabbitMQ 的基本架构:

- Broker:RabbitMQ服务器,负责接收和分发消息的应用。
- Virtual Host:虚拟主机,是RabbitMQ中的逻辑容器,用于隔离不同环境或不同应用程序的信息流。每个虚拟主机都有自己的队列、交换机等设置,可以理解为一个独立的RabbitMQ服务。
- Connection 连接:管理和维护与RabbitMQ服务器的TCP连接,生产者、消费者通过这个连接和 Broker 建立物理网络连接。
- Channel通道:是在Connection 内创建的轻量级通信通道,用于进行消息的传输和交互。应用程序通过Channel进行消息的发送和接收。通常一个 Connection 可以建立多个 Channel。
- Exchange交换机:交换机是消息的中转站,负责接收来自生产者的消息,并将其路由到一个或多个队列中。RabbitMQ 提供了多种不同类型的交换机,每种类型的交换机都有不同的消息路由规则。
- 直连交换机(Direct Exchange): 将消息路由到与消息中的路由键(Routing Key)完全匹配的队列。
- 主题交换机(Topic Exchange): 根据通配符匹配路由键,将消息路由到一个或多个队列。
- 扇出交换机(Fanout Exchange): 将消息广播到所有与交换机绑定的队列,忽略路由键。
- 头部交换机(Headers Exchange): 根据消息头中的属性进行匹配,将消息路由到与消息头匹配的队列。
- Queue队列:队列是消息的存储位置。每个队列都有一个唯一的名称。消息从交换机路由到队列,然后等待消费者来获取和处理。
- Binding绑定关系: Binding 是 Exchange 和 Queue 之间的关联规则,定义了消息如何从交换机路由到特定的队列。
此外,生产者和消费者也是RabbitMQ的核心组件,生产者负责发送消息到Exchange或者 Queue,消费者负责从Queue中订阅和处理消息。
这些核心组件共同构建了 RabbitMQ 的消息传递系统,他们协同工作才能实现消息的可靠传递、路由和业务处理等功能。
RabbitMQ 中有哪几种交换机类型?
详情可以看这篇文章:RabbitMQ入门
RabbitMQ 支持多种交换机(Exchange)类型,每种类型都用于不同的消息路由和分发策略:
Direct Exchange:
这种交换机根据消息的路由键(Routing Key)将消息发送到与之完全匹配的队列。只有当消息的路由键与队列绑定时指定的路由键完全相同时,消息才会被路由到队列。这是一种简单的路由策略,适用于点对点通信。Topic Exchange:
这种交换机根据消息的路由键与队列绑定时指定的路由键模式(通配符)匹配程度,将消息路由到一个或多个队列。路由键可以使用通配符符号 *(匹配一个单词)和 #(匹配零个或多个单词),允许更灵活的消息路由。用于发布/订阅模式和复杂的消息路由需求。Headers Exchange:
这种交换机根据消息的标头信息(Headers)来决定消息的路由,而不是使用路由键。队列和交换机之间的绑定规则是根据标头键值对来定义的,只有当消息的标头与绑定规则完全匹配时,消息才会被路由到队列。适用于需要复杂消息匹配的场景。Fanout Exchange:
这种交换机将消息广播到与之绑定的所有队列,无论消息的路由键是什么。用于发布/订阅模式,其中一个消息被广播给所有订阅者。Default Exchange:
这是 RabbitMQ 默认实现的一种交换机,它不需要手动创建。当消息发布到默认交换机时,路由键会被解释为队列的名称,消息会被路由到与路由键名称相同的队列。默认交换机通常用于点对点通信,但不支持复杂的路由策略。
RabbitMQ 支持哪些消息模式?
RabbitMQ 支持多种消息传递模式,这些模式允许应用程序在不同的场景下进行灵活的消息交流。以下是几种最常见的消息分发机制:
- workQueue 工作序列机制: Producer 将消息发送到 queue,多个 Consumer 同时消费Queue 上的消息。消息会均匀的分配给多个 Consumer 处理。
- Publish/Subscribe 订阅发布机制: Producer 只负责将消息发送到exchange交换机上。Exchange 将消息转发到所有订阅的 Queue,并由对应的 Consumer 去进行消费
- Routing 基于内容路由机制:在订阅发布机制的基础上,增加一个routingKey,并根据routingKey判断 Exchange 将消息转发到哪些 Queue 上。
- Topic 基于话题路由机制:在基于内容路由的基础上,对routingKey增加了模糊匹配的功能。
另外,RabbitMQ 还支持双向同步的 RPC 机制,不过一般用得比较少。这些消息模式允许开发者根据应用程序的需求选择合适的消息通信方式,来满足不同的业务场景和可靠性要求。不同的模式可以用于构建各种类型的分布式系统和应用程序。
RabbitMQ 如何实现消息的持久化?
RabbitMQ 允许消息的持久化,以确保即使在 RabbitMQ 服务器重新启动后,消息也不会丢失。RabbitMQ 可以通过以下方式实现消息的持久化:
- 消息持久化:在 RabbitMQ 中,只需要在发送消息时,将delivery_mode属性设置为 2,就可以将消息标记为持久化。
- 队列持久化:在 RabbitMQ 中声明队列时,也可以将队列声明为持久化。RabbitMQ 中的队列分为三种不同类型经典队列,仲裁队列和流式队列。其中,经典队列需要将durable属性设置为true。而仲裁队列和流式队列默认必须持久化保存。
- 交换机持久化:与经典队列类似,RabbitMQ 也可以在声明交换机时,将交换机的 durable 属性设置为true,这样就可以将交换机标记为持久化。
要注意,队列是可以被持久化,但是里面的消息是否为持久化那还要看消息的持久化设置。也就是说,重启之前 queue 里面如果还有未发出去的消息的话,重启之后,消息是否还存在队列里面就要取决于在发送消息时对消息的设置。 对于消息的可靠性来说,只需要设置队列的持久化和消息的持久化即可。exchange 的持久化并没有什么影响,但是,如果 exchange 不设置持久化的话,当 broker 服务重启之后,exchange 将不复存在,这样会导致消息发送者 producer 无法正常发送消息。
RabbitMQ 的持久化机制会对其性能产生影响。因此,需要根据具体的业务场景和需求来权衡是否需要持久化以及需要哪种类型的持久化。
RabbitMQ 是如何实现死信队列的?
死信队列是 RabbitMQ 提供的一种特殊序列,处理那些无法被正常消费的消息。有三种情况会产生死信:
- 消息被消费者明确拒绝。
- 消息达到预设的过期时间仍没有消费者消费。
- 消息由于队列已经达到最大长度限制而被丢弃。
在 RabbitMQ 中,实现死信队列只需要给正常队列增加三个核心参数即可:
- dead-letter-exchange:指定当前队列对应的死信队列
- dead-letter-routing-key:指定消息转入死信队列时的路由键
- message-ttl:消息在队列中的过期时间。
接下来,就可以往正常队列中发送消息。如果消息满足了某些条件,就会成为死信,并被重新发送到对应的死信队列中。而此时,RabbitMQ 会在消息的头部添加一些与死信相关的补充信息,例如时间、成为死信的原因、原队列等。应用程序可以按需处理这些补充的信息。
最后,死信队列中的消息都是正常业务处理失败的消息,应用程序需要创建一个消费者来专门处理这些被遗漏的消息。例如记录日志、发送警报等。这样才能保证业务数据的完整性。
RabbitMQ 中如何进行事务处理?
RabbitMQ 提供了事务处理机制,允许生产者在发送消息时将操作包装在一个事务中,以确保消息的可靠性传递。在 RabbitMQ 中,事务是通过通道(Channel)来实现的。可以通过以下步骤进行事务处理:
- 开启事务:在生产者端,可以通过调用 Channel 的 tx_select 方法来开启一个事务。这将启动一个新的事务,并将所有后续的消息发布操作放在该事务内。
- 发送消息:接下来在事务中,可以正常发送消息。如果消息发送失败,事务会自动回滚。
- 提交事务:如果事务中所有消息发送成功后,需要提交事务。可以通过调用 Channel 的tx_commit方法提交事务。
- 处理异常:如果在事务过程中发生异常,可以使用 try/catch 快来捕获异常。然后在异常处理过程中,调用 Channel 的 tx_rollback 方法来回滚 RabbitMQ 相关的事务操作。
需要注意的是,RabbitMQ 的事务处理是基于存储过程的,它可以保证在事务中的操作要么全部成功,要么全部失败。但是,由于 RabbitMQ 是一个异步的消息队列系统,事务处理可能会对其性能产生影响。因此,需要根据具体的应用场景和需求来权衡是否需要使用事务以及如何使用事务。
RabbitMQ 如何构建集群?
RabbitMQ 支持两种主要类型的集群:普通集群(Classic Cluster)和镜像集群(Mirrored Cluster)。他们之间有一些重要的区别:
- 普通集群: 这种模式使用Erlang语言天生具备的集群方式搭建。这种集群模式下,集群的各个节点之间只会有相同的元数据,即队列结构,而消息不会进行冗余,只存在一个节点中。消费时,如果消费的不是存有数据的节点, RabbitMQ会临时在节点之间进行数据传输,将消息从存有数据的节点传输到消费的节点。很显然,这种集群模式的消息可靠性不是很高。因为如果其中有个节点服务宕机了,那这个节点上的数据就无法消费了,需要等到这个节点服务恢复后才能消费,而这时,消费者端已经消费过的消息就有可能给不了服务端正确应答,服务起来后,就会再次消费这些消息,造成这部分消息重复消费。 另外,如果消息没有做持久化,重启就消息就会丢失。并且,这种集群模式也不支持高可用,即当某一个节点服务挂了后,需要手动重启服务,才能保证这一部分消息能正常消费。所以这种集群模式只适合一些对消息安全性不是很高的场景。而在使用这种模式时,消费者应该尽量的连接上每一个节点,减少消息在集群中的传输。
- 镜像集群:这种模式是在普通集群模式基础上的一种增强方案,这也就是RabbitMQ的官方HA高可用方案。需要在搭建了普通集群之后再补充搭建。其本质区别在于,这种模式会在镜像节点中间主动进行消息同步,而不是在客户端拉取消息时临时同步。并且在集群内部有一个算法会选举产生master和slave,当一个master挂了后,也会自动选出一个来。从而给整个集群提供高可用能力。这种模式的消息可靠性更高,因为每个节点上都存着全量的消息。而他的弊端也是明显的,集群内部的网络带宽会被这种同步通讯大量的消耗,进而降低整个集群的性能。这种模式下,队列数量最好不要过多
总的来说,普通集群适用于对性能要求高,但可以接受数据丢失的情况。而镜像集群则适用于对数据持久性和可用性有更高要求,并愿意付出一些性能代价的场景。
RabbitMQ如何保证消息不丢失
丢失原因分析

从上述流程我们可以得知:消息从生产者到达消费者,经过两次网络传输,并且在 RabbitMQ 服务器中进行路由。
因此我们能知道整个流程中可能会出现三种消息丢失场景:
- 生产者发送消息到 RabbitMQ 服务器的过程中出现消息丢失。 可能是网络波动未收到消息,又或者是服务器宕机。
- RabbitMQ 服务器消息持久化出现消息丢失。 消息发送到 RabbitMQ 之后,未能及时存储完成持久化,RabbitMQ 服务器出现宕机重启,消息出现丢失。
- 消费者拉取消息过程以及拿到消息后出现消息丢失。 消费者从 RabbitMQ 服务器获取到消息过程出现网络波动等问题可能出现消息丢失;消费者拿到消息后但是消费者未能正常消费,导致丢失,可能是消费者出现处理异常又或者是消费者宕机。
针对上述三种消息丢失场景,RabbitMQ 提供了相应的解决方案,confirm 消息确认机制(生产者),消息持久化机制(RabbitMQ 服务),ACK 事务机制(消费者)
接下来看解决方案
confirm 消息确认机制(生产者)
Confirm 模式是 RabbitMQ 提供的一种消息可靠性保障机制。当生产者通过 Confirm 模式发送消息时,它会等待 RabbitMQ 的确认,确保消息已经被正确地投递到了指定的 Exchange 中。
- 消息正确投递到 queue 时,会返回 ack。
- 消息没有正确投递到 queue 时,会返回 nack。如果 exchange 没有绑定 queue,也会出现消息丢失。
使用方法:
- 生产者通过 confirm.select 方法将 Channel 设置为 Confirm 模式。
- 发送消息后,通过添加 add_confirm_listener 方法,监听消息的确认状态。
消息持久化机制(RabbitMQ 服务)
持久化机制是指将消息存储到磁盘,以保证在 RabbitMQ 服务器宕机或重启时,消息不会丢失。
使用方法:
- 生产者通过将消息的 delivery_mode 属性设置为 2,将消息标记为持久化。
- 队列也需要进行持久化设置,确保队列在 RabbitMQ 服务器重启后仍然存在。经典队列需要将durable属性设置为true。而仲裁队列和流式队列默认必须持久化保存。
注意事项: 持久化机制会影响性能,因此在需要确保消息不丢失的场景下使用。
ACK 事务机制
用于确保消息被正确消费。当消息被消费者成功处理后,消费者发送确认(ACK)给 RabbitMQ,告知消息可以被移除。这个过程是自动处理的,也可以关闭进行手工发送 ACK。
使用方法:
- 在 RabbitMQ 中,ACK 机制默认是开启的。当消息被消费者接收后,会立即从队列中删除,除非消费者发生异常。
- 可以手动开启 ACK 机制,通过将 auto_ack 参数设置为 False,手动控制消息的 ACK。
注意事项: ACK 机制可以确保消息不会被重复处理,但如果消费者发生异常或者未发送 ACK,消息可能会被重复投递。
RabbitMQ中如何解决消息堆积问题
消息堆积原因

接下来看解决方案
消费者处理消息的速度太慢
- 增加消费者数量:通过水平扩展,增加消费者的数量来提高处理能力。
- 优化消费者性能:提高消费者处理消息的效率,例如优化代码、增加资源。
- 消息预取限制(prefetch count):调整消费者的预取数量以避免一次处理过多消息而导致处理缓慢。
队列的容量太小
增加队列的容量:调整队列设置以允许更多消息存储。
网络故障
- 监控和告警:通过监控网络状况并设置告警,确保在网络故障时快速发现并解决问题。
- 持久化和高可用性:确保消息和队列的持久化以避免消息丢失,并使用镜像队列提高可用性。
消费者故障
- 使用死信队列:将无法处理的消息转移到死信队列,防止堵塞主队列。
- 容错机制:实现消费者的自动重启和错误处理逻辑。
队列配置不当
优化队列配置:检查并优化消息确认模式、队列长度限制和其他相关配置。
消息大小
消息分片:将大型消息分割成小的消息片段,加快处理速度。
业务逻辑复杂或耗时
优化业务逻辑:简化消费者中的业务逻辑,减少处理每个消息所需的时间。
消息产生速度快于消费速度
- 使用消息限流:控制消息的生产速度,确保它不会超过消费者的处理能力。
- 负载均衡:确保消息在消费者之间公平分配,避免个别消费者过载。
其他配置优化
- 消息优先级:使用消息优先级确保高优先级消息优先处理。
- 调整RabbitMQ配置:优化RabbitMQ服务的配置,如文件描述符限制、内存使用限制等。
RabbitMQ中如何保证消息不被重复消费
什么情况会导致消息被重复消费呢
- 生产者:生产者可能会重复推送一条数据到 MQ 中,比如 Controller 接口被重复调用了 2 次,没有做接口幂等性导致的;
- MQ:在消费者消费完准备响应 ack 消息消费成功时,MQ 突然挂了,导致 MQ 以为消费者还未消费该条数据,MQ 恢复后再次推送了该条消息,导致了重复消费。
- 消费者:消费者已经消费完消息,正准备但是还未响应给ack消息到时,此时消费者挂了,服务重启后 MQ 以为消费者还没有消费该消息,再次推送了该条消息。

接下来看解决方案
使用数据库唯一键约束
缺点:局限性很大,仅仅只能用在我们数据新增场景,并且性能也比较低
使用乐观锁
假设是更新订单状态,在发送的消息的时候带上修改字段的版本号
缺点:如果说更新字段比较多,并且更新场景比较多,可能会导致数据库字段增加并且还有可能出现多条消息同时在队列中此时他们修改字段版本号一致,排在后续的消息无法被消费
简单的消息去重,插入消费记录,增加数据库判断

优点:很多场景下的确能起到不错的效果
缺点:
- 这个消费者的代码执行需要1秒,重复消息在执行期间(假设100毫秒)内到达(例如生产者快速重发,Broker重启等),增加校验的地方是不是还是没数据(因为上一条消息还没消费完,没有记录)
- 那么就会穿透掉检查的挡板,最后导致重复的消息消费逻辑进入到非幂等安全的业务代码中,从而引发重复消费的问题
并发消息去重基于消息幂等表

缺点:如果说第一次消息投递异常没有消费成功,并且没有将消息状态给置为成功或者没有删除消息表记录,此时延时消费每次执行下列都是一直处于消费中,最后消费就会被视为消费失败而被投递到死信Topic中
方案:插入的消息表必须要带一个最长消费过期时间,例如10分钟

上述方案只需要一个存储的中心媒介,那我们可以选择更灵活的存储中心媒介,比如Redis。使用Redis有两个好处:
- 性能上损耗更低
- 上面我们讲到的超时时间可以直接利用Redis本身的ttl实现
总结
- 利用数据库唯一键约束
- 可以利用乐观锁
- 插入消费记录
不丢和不重是矛盾的(在分布式场景下),总的来说,开发者根据业务的实际需求来选择相应的方式即可。
RocketMQ
RocketMQ消费消息是推模式还是拉模式
RocketMQ实际上同时支持推模式和拉模式来消费消息,但这里有一个有趣的细节:虽然RocketMQ提供了所谓的"推模式"消费者,但在底层实现上,它仍然是基于拉模式的。让我们详细探讨这两种模式,并通过代码示例来解释它们的工作原理。
- 推模式(Push Mode)
虽然称为"推模式",但RocketMQ的DefaultMQPushConsumer实际上是在内部通过长轮询(Long Polling)来实现的,这本质上还是一种拉模式。不过,对于开发者来说,使用起来就像是推模式,因为消息会自动被"推送"到消费者的监听器中。
- 拉模式(Pull Mode)
拉模式允许消费者主动从Broker拉取消息。这种模式给了开发者更多的控制权,可以根据自己的节奏来消费消息。
比较和选择
推模式(Push):
- 优点:使用简单,自动管理消费进度,适合大多数场景。
- 缺点:灵活性较低,无法完全控制消费节奏。
拉模式(Pull):
- 优点:更灵活,可以精确控制消费速度和批次大小。
- 缺点:需要自己管理偏移量,实现相对复杂。
虽然RocketMQ提供了所谓的"推模式",但实际上它是基于拉模式实现的。对于大多数应用场景,使用DefaultMQPushConsumer就足够了,因为它提供了一个类似推模式的简单接口。只有在需要精细控制消息消费过程时,才需要考虑使用拉模式。
RocketMQ 的广播消息和集群消息有什么区别?
广播消息和集群消息是 RocketMQ 的两种不同的消息消费模式。其中
- 广播模式意味着一条消息会被发送到所有订阅了这个主题 Topic 的消费者,而所有消费者都会收到相同的消息副本。
- 集群模式意味着一条消息只会分发给订阅了这个主题 Topic 的同一个消费者组中的一个消费者处理。每个消费者组只会处理一次消息。
他们的区别主要提现在实现方式以及适用场景上。
- 集群模式在 Broker 端统一管理每个消费者组的消费进度,对消费进度的管理是严格的。这样,每次消费者服务启动后,都可以从上一次消费的进度开始开始进行消费。 而广播模式是交由每个消费者自行管理消费进度,消费进度的管理是不严格的,容易产生丢失。当消费者服务启动后,如果本地的消费进度丢失了,就只能消费到启动之后的消息,而无法从上一次消费的进度开始消费。因此,广播模式对于消息的连续性保证是不强的。
- 集群模式适用于大多数常规对消息安全敏感的业务场景,例如订单处理、库存管理等。多个消费者协同工作可以提高消息的处理能力并实现消息的负载均衡。而广播模式适用于一些对消息安全不太敏感的特殊业务场景。例如日记记录、时间通知等。这些场景下所有的消费者都需要处理相同的消息。
RocketMQ如何实现消息的持久化
消息存储的主要组件包括 CommitLog 文件、消费队列文件(ConsumeQueue)、以及索引文件(IndexFile)。
- CommitLog文件 :CommitLog是RocketMQ的核心存储文件,负责保存消息的完整内容。
- 顺序写入:所有的消息都顺序写入CommitLog文件,这种方式减少了磁盘寻道时间,提高了写入性能。
- 文件滚动:CommitLog按照固定大小(比如1GB)进行分片。当一个文件写满后,会创建一个新的文件。
- 存储所有数据:包括消息体、主题、队列ID等。
- ConsumeQueue文件 :ConsumeQueue是针对消息的逻辑视图,旨在加快消费者对消息的访问速度。
- 条目固定:每个ConsumeQueue条目固定为20字节,包含消息在CommitLog中的偏移量、消息大小、Tag哈希值。
- 独立文件:每个主题的每个队列都有独立的ConsumeQueue文件,文件路径为store/consumequeue/{topic}/{queueId}。
- 快速定位:通过ConsumeQueue,消费者无需扫描整个CommitLog即可快速找到消息的位置。
- IndexFile文件 :IndexFile用于支持消息的快速检索。
- 哈希索引:为消息的key建立哈希索引,支持通过key快速检索消息偏移。
- 增强查询:IndexFile是可选的,用于需要基于消息属性进行快速查找的场景。
消息存储流程
- 接收消息:Broker接收到消息后,将其放入内存缓冲区(待写入CommitLog)。
- 写入CommitLog: 每条消息追加到当前活跃的CommitLog文件中。使用顺序写入提升写入效率和磁盘利用率。
- 同步到ConsumeQueue:
- 异步转发服务(ReputMessageService)从CommitLog读取新写入的消息。
- 将消息的偏移量和其他元数据(如大小和Tag哈希值)存储到相应的ConsumeQueue文件中。
- 更新IndexFile(可选): 若消息带有key(如业务ID),则将其哈希和偏移量存入IndexFile。 这样,可以通过该key快速查找消息。
RocketMQ的消息存储如何进行清理和归档
RocketMQ 提供了消息存储清理和归档的机制,以便管理消息存储空间,删除过期消息,并将历史消息归档到其他存储介质中。这些功能有助于维护消息队列的性能和可用性。以下是关于 RocketMQ 消息存储清理和归档的主要方面:
- 消息文件删除策略: RocketMQ 支持多种消息文件删除策略,可以在配置文件中进行设置。以下是一些常见的策略:
- 定时删除策略: 您可以配置 RocketMQ 定期删除过期的消息文件和索引文件。这样,一旦消息文件中的消息过期,RocketMQ 将自动删除它们。
- 空间满策略: 如果存储磁盘空间达到一定限制,RocketMQ 可以自动删除最早的消息文件,以释放磁盘空间。这个策略确保了存储空间不会无限制地增长。
- 指定时间段删除策略: 您可以配置 RocketMQ 只删除特定时间段内的消息文件,以保留历史消息。
- 消息归档: RocketMQ 允许您将历史消息归档到其他存储介质中,以减小消息服务器的存储负担。归档通常涉及将消息转移到长期存储(如云存储或本地归档系统)中。归档可以手动触发,也可以自动触发,具体取决于您的需求。
- 历史消息访问: 尽管消息被归档,RocketMQ 仍然提供了访问历史消息的机制。通过合适的归档系统或者存储介质,您可以检索和访问历史消息,以满足合规性要求或其他业务需求。
需要注意的是,清理和归档消息不是 RocketMQ 的核心功能,而是辅助功能。需要根据自己的需求和业务场景来确保配置合理的清理策略以防止存储空间耗尽,并根据业务需求进行消息的归档操作,以保留历史消息数据。同时,归档后的消息可以根据需要进行合适的检索和恢复,以满足特定的数据需求。
RocketMQ延迟消息是如何实现的
RocketMQ通过特定的延迟级别设计实现延迟消息功能。在RocketMQ中,延迟消息是通过设置消息的延迟级别(Delay Level)来实现的。每个延迟级别对应一个特定的时间段,这样可以让消息在指定的时间之后才被消费。
实现原理
- 延迟级别:RocketMQ不支持任意时间的延迟,而是提供了18个固定的延迟级别,从1s,5s,10s,30s,1m,2m,3m到2h不等。
- 2特殊主题:所有延迟消息都会先发送到一个特殊的内部主题 SCHEDULE_TOPIC_XXXX。
- 定时任务:Broker会启动一个定时任务,按照延迟时间的先后顺序依次扫描每个延迟级别队列。
- 消息转移:当扫描到期的消息时,会将消息从延迟队列转移到目标主题。
- 消费者消费:消息被转移到目标主题后,消费者就可以正常消费这条消息了。
RocketMQ提供了哪些消息过滤机制
RocketMQ提供了多种消息过滤机制,以便消费者能够根据业务需求进行精确的消息消费。主要的过滤机制包括:
- Tag过滤:最常用的过滤方式,消费者可以基于消息的Tag进行过滤。
- SQL92语法过滤:基于Message属性,以SQL92标准语法进行复杂条件过滤,该功能需要Broker支持。
RocketMQ事务消息是如何实现的
RocketMQ的事务消息对分布式系统中的事务一致性问题提供了有效的解决方案。比如当用户下单时,需要同时处理订单生成、库存扣减、支付处理等操作。事务消息确保所有这些操作要么全部成功,要么全部回滚。
RocketMQ的事务消息实现是一个复杂而精巧的过程,主要通过两阶段提交(2PC)和补偿机制来保证分布式事务的一致性。
下面我将结合下单减库存场景详细解释,这里订单和库存是两个微服务,所以这两步操作存在分布式事务问题,我们可以在下单操作完成发送减库存事务消息异步扣减库存,使用 RocketMQ 的事务消息,可以确保订单创建和库存扣减的事务一致性。以下是实现该场景的详细步骤:
假设我们有两个服务:
- 订单服务(Order Service):负责处理订单创建。
- 库存服务(Inventory Service):负责管理库存更新。
使用事务消息实现步骤
- 发送半消息: 在订单服务中,首先发送一个库存扣减的半消息到 RocketMQ。这条消息在事务未完成前,对消费者(库存服务)是不可见的。
- 执行本地事务(创建订单): 在订单服务中执行本地事务,即在数据库中创建订单记录。
- 根据本地事务结果提交或回滚事务消息:
- 如果订单创建成功,提交事务消息,使库存扣减消息对库存服务可见。
- 如果订单创建失败,回滚事务消息,库存不变。
- 处理MQ事务消息状态回查: 如果RocketMQ未收到事务提交或回滚的结果,会定期询问订单服务,以确认事务的最终状态。
- 原子性:半事务消息对消费者不可见,确保了事务的原子性。
- 持久性:半事务消息会被持久化,即使Broker宕机也能恢复。
- 隔离性:通过特殊的队列存储半事务消息,实现了隔离。
- 一致性:通过二阶段提交和回查机制保证了最终一致性。
通过这套实现机制,RocketMQ能够在分布式系统中实现可靠的事务消息,确保消息发送与本地事务的一致性,即使在系统崩溃或网络故障的情况下也能保证数据的一致性。
RocketMQ的集群架构是怎样的
RocketMQ的集群架构设计旨在提高系统的可用性、可靠性和可扩展性。它通过多种组件协同工作,实现消息的生产、存储、分发和消费。以下是关于RocketMQ集群架构及其使用场景的详细说明。
RocketMQ集群架构
NameServer:
- 它是一个几乎无状态的节点,可以集群部署用于服务发现。
- NameServer为Producer和Consumer提供路由信息。
Broker:
- Broker负责接收、存储和转发消息。
- 它可以分为Master和Slave,支持多对Master-Slave配置以实现高可用。
- Master和Slave之间通过同步/异步机制进行数据复制。
Producer:
- 消息生产者负责产生消息,并发送到Broker。
- 支持同步和异步发送消息。
Consumer:
- 消息消费者负责从Broker接收消息。
- 支持Push和Pull两种消费方式,可以是集群消费或广播消费。
使用场景
- 高可用性:通过Master-Slave配置和Broker高可用机制,保证系统在部分节点失效的情况下仍能正常运行。
- 消息持久化:控制消息的存储策略(同步/异步),从而达成更高的可靠性需求。
- 负载均衡:使用多NameServer和Broker集群,支持负载均衡,从而支持大规模消息流量。
- 弹性扩展:可以按需添加NameServer和Broker节点,进行水平扩展。
RocketMQ的Broker有哪几种集群模式
RocketMQ的Broker有三种集群模式:
- 单Master模式:只有一个Master节点,其他都是Slave节点。Master节点负责响应客户端的请求并存储消息,Slave节点只同步Master节点的消息,也会响应部分客户端的读请求。这种模式的优点是简单易部署,但是存在单点故障的问题,如果Master节点宕机,会导致整个服务不可用。
- Master-Slave模式(经典双集群部署):一个Master节点对应多个Slave节点,Master和Slave都是独立的NameServer。Master节点负责响应客户端请求并存储消息,Slave节点只同步Master节点的消息,也会响应部分客户端的读请求。这种模式的优点是高可用性,即使Master节点宕机,Slave节点可以自动升级为Master节点,继续提供服务。但是,如果只有一个Master节点,存在单点故障的问题。
- Dledger模式(高可用集群部署):在Master-Slave模式的基础上增加了Raft协议,实现了自动脑裂后的数据高可靠性。即使某个节点从网络上掉下来或者宕机后,仍然能够保证所有的消息不会丢失。这种模式的优点是高可用性和高可靠性,即使某个节点出现故障,也能保证服务的可用性。
- 多Master多Slave模式(异步复制):每个Master节点配置一个Slave节点,有多对Master-Slave,HA采用异步复制方式。这种模式下,即使磁盘损坏,消息丢失的非常少,且消息实时性不会受影响。Master宕机后,消费者仍然可以从Slave消费,过程对应用透明,不需要人工干预,性能与多Master模式几乎一样。缺点是Master宕机或磁盘损坏情况下可能会丢失少量消息
- 多Master多Slave模式(同步双写):每个Master节点配置一个Slave节点,有多对Master-Slave,HA采用同步双写方式,即只有主备都写成功,才向应用返回成功。这种模式的优点是数据与服务都无单点故障,Master宕机情况下,消息无延迟,服务可用性与数据可用性都非常高。缺点是性能比异步复制模式略低,发送单个消息的响应时间会略高
总的来说,单Master模式适合测试和开发环境,Master-Slave模式适合生产环境,而Dledger模式适合需要高可靠性的生产环境。
RocketMQ如何保证消息顺序
RocketMQ提供了两种级别的顺序消息:全局顺序消息和分区顺序消息。
- 全局顺序消息:全局顺序消息确保一个主题内的所有消息都按照发送顺序被消费。这通常通过将所有消息路由到同一个队列来实现。
- 分区顺序消息:分区顺序消息保证具有相同分区键的消息按顺序被消费。这允许更高的并行度,因为不同分区键的消息可以并行处理。
注意事项
- 全局顺序消息可能会限制系统的吞吐量,因为所有消息都经过单一队列。
- 分区顺序消息在保证局部顺序的同时提供了更好的并行性。
- 选择合适的分区键对于分区顺序消息至关重要,以确保相关消息进入同一队列。
- 消费者端需要正确处理并发和重试逻辑,以维护消息顺序。
RocketMQ如何保证消息不丢失
RocketMQ通过多层面的机制来确保消息的可靠性,包括生产者端、broker端和消费者端。
生产者端保证
- 同步发送 :同步发送是最可靠的发送方式,它会等待broker的确认响应。
- 异步发送 + 重试机制 :异步发送通过回调来处理发送结果,并可以设置重试次数。
Broker端保证
- 同步刷盘 :通过配置broker.conf文件,可以启用同步刷盘:flushDiskType = SYNC_FLUSH
- 主从复制 :配置主从架构,并设置同步复制:brokerRole = SYNC_MASTER
消费者端保证
- 手动提交消费位移 :使用手动提交可以确保消息被正确处理后再提交位移。
- 幂等性消费 :在消费端实现幂等性处理,确保重复消费不会导致业务问题。
RocketMQ如何解决消息积压问题
消息积压是消息中间件中常见的问题,主要由消费速度跟不上生产速度导致。以下是几种解决方案:
- 增加消费者线程数量:这是最直接的方法,通过增加消费者线程数来提高消费能力。
- 消息业务异步处理
- 调整消费者的消费模式:将顺序消费改为并行消费,提高消费效率。
- 使用消息过滤 :通过消息过滤,只消费重要的消息,降低消费压力。
- 调整生产者发送策略:如果可能,可以调整生产者的发送策略,如降低发送频率或者实现背压机制。
这些方法可以单独使用,也可以组合使用,具体取决于具体的业务场景和系统架构。在实施这些解决方案时,请注意监控系统性能,确保不会因为过度优化而导致其他问题。
Kafka
Kafka的优势和特点
优势:
- 高吞吐量:单机每秒处理几十上百万的消息量。即使存储了许多TB的消息,它也保持稳定的性能。
- 高性能:单节点支持上千个客户端,并保证零停机和零数据丢失,异步化处理机制
- 持久化:将消息持久化到磁盘。通过将数据持久化到硬盘以及replica(follower节点)防止数据丢失。
- 零拷贝:减少了很多的拷贝技术,以及可以总体减少阻塞事件,提高吞吐量。
- 可靠性 :Kafka是分布式,分区,复制和容错的。
Kafka的特点 :
- 顺序读,顺序写
- 利用Linux的页缓存
- 分布式系统,易于向外扩展。所有的Producer、Broker和Consumer都会有多个,均为分布式的。无需停机即可扩展机器。多个Producer、Consumer可能是不同的应用。
- 客户端状态维护:消息被处理的状态是在Consumer端维护,而不是由server端维护。当失败时能自动平衡。
- 支持online(在线)和offline(离线)的场景。
- 支持多种客户端语言。Kafka支持Java、.NET、PHP、Python等多种语言。
Kafka与其它消息队列的对比

各种对比之后,有如下建议:
- ActiveMQ,没经过大规模吞吐量场景的验证,社区也不是很活跃,所以不推荐;
- RabbitMQ,虽然erlang 语言阻止了大量的 Java 工程师去深入研究和掌控它,对公司而言,几乎处于不可控的状态,但是毕竟是开源的,比较稳定的支持,活跃度也高,推荐中小型公司使用;推荐
- RocketMQ,阿里出品,Java语言编写,经过了阿里多年双十一大促的考验,性能和稳定性得到了充分的严重。目前在业界被广泛应用在订单,交易,充值,流计算,消息推送,日志流式处理,binlog分发等场景;强烈推荐
- Kafka,如果是大数据领域的实时计算、日志采集等场景,用 Kafka 是业内标准的,绝对没问题,社区活跃度很高,绝对不会黄,何况几乎是全世界这个领域的事实性规范。
Kafka与RabbitMQ相比有什么优势?
Kafka 和 RabbitMQ 都是流行的消息中间件系统,他们各自都有一些优势和适用场景。以下是 Kafka 相对于 RabbitMQ 的一些比较明显的优势:
- 分布式架构: Kafka 是为大规模分布式流处理而设计的,具有高度可伸缩性。RabbitMQ 虽然也支持分布式架构,但相对而言,Kafka 的集群设计更完善,更适合处理大规模的消息流。
- 吞吐量: Kafka每秒可处理十几万消息,而 RabbitMQ 每秒可处理几万条消息。
- 消息复制和可用性:Kafka 允许配置多个消息副本,确保数据的冗余存储,提高可用性和容错性。RabbitMQ 也支持镜像队列以实现冗余,但是不如 Kafka 的多副本复制灵活。
- 时间溯源:Kafka 在事件溯源和事件驱动架构中非常强大。他允许事件在 Topic 中保留一段时间,以便后续的分析和回溯查询。RabbitMQ 通常用于实时消息传递,对于事件溯源不够灵活。
- 批处理和流处理: Kafka 提供了流处理 API,可用于实时数据流处理等场景。而 RabbitMQ 倾向于更专注的处理实时消息传递。
- 社区和生态系统:Kafka 有一个庞大的社区和丰富的生态系统,提供了许多与大数据和流处理相关的工具和库。RabbitMQ 也有一个活跃的社区,但是相对而言社区规模以及社区活跃性就要小很多。
如果需要处理大规模的实时数据流或事件驱动架构,Kafka 可能更适合;如果更关注传统的消息传递和队列处理,RabbitMQ 的高级功能更丰富,可能更合适。因此,选择哪种消息中间件还是要取决于具体的应用场景。
Kafka有哪些组件
kafka运行在集群上,集群包含一个或多个服务器。kafka把消息存在topic中,每一条消息包含键值(key),值(value)和时间戳(timestamp)。
kafka有以下一些基本概念:
- Producer - 消息生产者,就是向kafka broker发消息的客户端。
- Consumer - 消息消费者,是消息的使用方,负责消费Kafka服务器上的消息。
- Topic - 主题,由用户定义并配置在Kafka服务器,用于建立Producer和Consumer之间的订阅关系。生产者发送消息到指定的Topic下,消息者从这个Topic下消费消息。
- Partition - 消息分区,一个topic可以分为多个 partition,每个partition是一个有序的队列。partition中的每条消息都会被分配一个有序的id(offset)。
- Broker - 一台kafka服务器就是一个broker。一个集群由多个broker组成。一个broker可以容纳多个topic。
- Consumer Group - 消费者分组,用于归组同类消费者。每个consumer属于一个特定的consumer group,多个消费者可以共同消息一个Topic下的消息,每个消费者消费其中的部分消息,这些消费者就组成了一个分组,拥有同一个分组名称,通常也被称为消费者集群。
- Offset - 消息在partition中的偏移量。每一条消息在partition都有唯一的偏移量,消息者可以指定偏移量来指定要消费的消息。
Kafka工作流程

- producer先从zookeeper的 "/brokers/.../state"节点找到该partition的leader
- producer将消息发送给该leader
- leader将消息写入本地log
- followers从leader pull消息
- 写入本地log后向leader发送ACK
- leader收到所有ISR中的replication的ACK后,增加HW(high watermark,最后commit 的offset)并向producer发送ACK
Kafka中的Topic和Partition有什么关系?
在Kafka中,Topic和Partition是两个密切相关的概念。
- Topic是Kafka中消息的逻辑分类,可以看作是一个消息的存储类别。它是按照不同的主题对消息进行分类,并且可以用于区分和筛选数据。每个Topic可以有多个Partition,每个Partition都是Topic的一个子集,包含了一部分特定的消息。
- Partition则是Kafka 中实际保存数据的单位。每个Topic可以被划分为多个Partition,而这些 Partition 会尽量平均的分配到各个 Broker 上。当一条消息发送到Kafka时,它会被分配到一个特定的Partition中,并最终写入 Partition 对应的日志文件里。这个分配过程是根据Partition的规则来完成的,比如可以按照消息的某个属性进行哈希或者按照时间戳进行排序等。
因此,Topic和Partition的关系是,Topic是消息的逻辑分类,用于区分和筛选数据,而Partition则是Topic的物理划分,用于将消息分配到不同的部分中以便于处理和存储。Topic 和 Partition 的设计对于高吞吐量和横向扩展非常有用。因为生产者和消费者只需要根据 Topic 进行具体的业务实现,而不用关心消息在集群内的分布情况。而在集群内部,这些 Partition 会尽量平均的分布在不同的 Broker节点上,从而提高了系统 整体的性能和可伸缩性。
Kafka的消费消息是如何传递的?
在Kafka中,消息的传递主要涉及三个环节:生产者生产消息、broker保存消息和消费者消费消息。
- 生产者生产消息:生产者负责将消息发布到Kafka broker。在发布消息时,生产者需要指定目标主题。消息被写入后,将被存储在指定分区的当前副本中。当发送消息失败时,生产者还会提供确认以及重试机制,以保证消息能够正确的发送到 Broker 上。
- broker保存消息:Kafka broker接收到生产者发送的消息后,会将其存储在内部的缓冲区中,等待消费者拉取。当消费者向broker发送拉取请求时,broker会从缓冲区中获取消息并返回给消费者。Kafka broker能够保证消息的可靠性和顺序性,即使在异常情况下(如服务器崩溃),也能够保证消息不会丢失。
- 消费者消费消息:消费者从Kafka broker中订阅指定的主题,并拉取消息进行消费。消费者可以以同步或异步的方式拉取消息,并对拉取到的消息进行处理。当消费者处理完消息后,会向Kafka broker发送确认消息,表示消息已经被成功处理。这样可以保证消息被正确处理且不会重复消费。
总体来说,Kafka通过生产者、Kafka broker和消费者的协同工作,实现了高吞吐量、高可靠性和高可扩展性的消息传递。
Kafka中的消息如何分配给不同的消费者?
Kafka中的消息是通过分区(Partition)分配给不同的消费者的。Kafka将每个Topic划分为多个Partition,每个Partition存储一部分消息。消费者通过订阅Topic来消费消息,而Kafka将Partition中的消息按照一定的分配策略分配给消费者组中的不同消费者。
Kafka提供了多种分区分配策略,用于确定如何将分区分配给消费者。例如:
- RoundRobin 轮询策略:Kafka将Partition按照轮询的方式分配给消费者组中的不同消费者,每个消费者依次获得一个Partition,直到所有Partition被分配完毕。当消费者数量发生变化时,Kafka会重新分配Partition。
- Range 范围策略:Kafka将Partition按照Range的方式分配给消费者组中的不同消费者,每个消费者负责处理指定范围内的Partition。这种分配方式适用于Topic的Partition数量较少,而消费者数量较多的情况。
- Sticky 粘性策略: 尽量保持每个消费者在一段时间内消费相同的分区,以减少分区重新分配的频率
当消费者处理完一个Partition中的所有消息后,它会向Kafka发送心跳请求,Kafka会将该Partition分配给其他消费者进行处理。这种机制确保了消息在不同的消费者之间负载均衡,并提高了容错性。如果一个消费者出现故障,其他消费者可以继续处理Partition中的消息,而不会导致消息丢失或重复处理。
Kafka中的消息是如何存储的?
Kafka 中的消息是以文件的方式持久化到磁盘中进行存储的,这是 Kafka 的一个关键特性,确保消息的可靠性和可用性。Kafka中的消息是通过以下方式进行存储的:
- Partition 分区:Partition是Kafka中消息存储的基本单位,每个Topic下的消息都会被划分成多个Partition进行管理。每个Partition都是一个有序的、不变的消息队列,消息按照追加的顺序被添加到队列尾部。
- Segment 分块:Partition会被进一步划分成多个Segment,Segment是逻辑上的文件组,方便进行数据的管理和查找。每个Segment里都包含多个文件,这些文件名相同且被集合在一起。
- 文件索引:Segment中的每个文件都有自己的索引文件和数据文件,索引文件存储了当前数据文件的索引信息,而数据文件则存储了当前索引文件名对应的数据信息。
- 消息偏移:Kafka中的每个消息都会被分配到一个特定的Partition中,然后根据Partition内的Segment划分,被存储到对应的数据文件中。消息的偏移量信息则会被记录在索引文件中。
- 持久化:Kafka中的每个消息都包含一个64位的偏移量,该偏移量表示消息在Partition中的位置。当消费者读取消息时,可以通过偏移量信息来确定需要从哪个位置开始读取。
Kafka 的消息存储是基于日志文件和分区的,确保了消息的可靠性、持久性和高吞吐量。消息被追加到日志文件中,每个消息都有唯一的偏移量,分区和副本机制保证了数据的冗余存储和可用性。这种设计使 Kafka 成为一个可信赖的消息传递系统,适用于各种实时数据处理、日志聚合和事件驱动应用程序。
Kafka如何保证消息可靠?
为了保证消息在传递过程当中,消息不会丢失或者被重复传递,Kafka 设计了非常多的重要机制来保证消息的可靠性。例如
- 数据冗余:Kafka通过将消息副本(replica)的方式来实现数据冗余,每个topic都可以配置副本数量,副本数量越多,数据可靠性越高,但会占用更多的存储空间和网络带宽。在 Kafka 中,针对每个 Partition,会选举产生一个 Leader 节点,负责响应客户端的请求,并优先保存消息。而其他节点则作为 Follower 节点,负责备份 Master 节点上的消息。
- 消息发送确认机制:Kafka支持对生产者发送过来的数据进行校验,以检查数据的完整性。可以通过设置生产者端的参数(例如:acks)来配置校验方式。配置为 0,则不校验生产者发送的消息是否写入 Broker。配置为 1,则只要消息在 Leader 节点上写入成功后就向生产者返回确认信息。配置为-1 或 all,则会等所有 Broker 节点上写入完成后才向生产者返回确认信息。
- ISR机制:针对每个 Partition,Kafka 会维护一个 ISR 列表,里面记录当前处于同步状态的所有Partition。并通过 ISR 机制确保消息不会在Master 故障时丢失。
- 消息持久化:Kafka将消息写入到磁盘上,而不是仅在内存中缓存。这样可以保证即使在系统崩溃的情况下,消息也不会丢失。并且使用零拷贝技术提高消息持久化的性能。
- 消费者确认机制:Kafka消费者在处理完消息后会向Kafka broker发送确认消息,表示消息已经被成功处理。如果消费者未发送确认消息,则Kafka broker会保留消息并等待消费者再次拉取。这样可以保证消息被正确处理且不会重复消费。
这些机制的组合确保了 Kafka 中消息的高可靠性和持久性,使得 Kafka 成为可靠的消息传递系统,适用于各种实时数据处理和日志聚合需求。
什么是ISR 机制?
Kafka根据副本同步的情况,分成了3个集合:
- AR(Assigned Replicas):包括ISR和OSR
- ISR(In-sync Replicas):和leader副本保持同步的副本集合,可以被认为是可靠的数据
- OSR(Out-Sync Replicas):和Leader副本同步失效的副本集合
当 kafka 副本同步机制是所有follower都同步成功才返回 ack 给生产者时,如果有一个follower,因为某种故障,迟迟不能与leader 进行同步,那leader 就要一直等下去,直到它完成同步,才能发送ack。这个问题怎么解决呢?
Leader维护了一个动态的in-sync replica set (ISR-同步副本列表),意为和leader保持同步的follower集合。根据follower发来的FETCH请求中的fetch offset判断ISR中的follower完成数据同步是否成功。如果follower长时间未向leader同步数据,则该follower将被踢出ISR,该时间阈值由replica.lag.time.max.ms参数设定。Leader发生故障之后,就会从ISR中选举新的leader。
- ISR(In-Sync Replicas ):与leader保持同步的follower集合
- AR(Assigned Replicas):分区的所有副本
- ISR是由leader维护,follower从leader同步数据有一些延迟(包括延迟时间replica.lag.time.max.ms和延迟条数replica.lag.max.messages两个维度, 当前最新的版本0.10.x中只支持replica.lag.time.max.ms这个维度),任意一个超过阈值都会把follower剔除出ISR, 存入OSR(Outof-Sync Replicas)列表,新加入的follower也会先存放在OSR中。
- AR=ISR+OSR。
如何确保Kafka集群的高可用?
Kafka设计了多种机制,共同保证集群的高可用性:
- 分布式架构:Kafka集群通常由多个Broker组成,每个Broker存储部分数据副本。这样,即使某个Broker出现故障,其他Broker也可以继续处理和存储消息,从而保证整体的高可用性。
- 数据冗余:Kafka通过数据冗余来保证高可用性。每个Topic的数据会被分成多个Partition,并在多个Broker上进行复制。即使某个Broker出现故障,数据仍然可以从其他Broker中获取。
- 副本机制:副本是Kafka实现高可用性的重要手段。Kafka中的每个Partition都有多个副本,这些副本分布在不同的Broker上,从而在部分Broker故障时,仍然有足够的副本可用以保证高可用性。
- 分区领导者选举:在Kafka中,每个Partition都有一个领导者(Leader)和零个或多个追随者(Follower)。当领导者不可用时,追随者会进行领导者选举,以保证系统的可用性。
- 消费者组实现负载均衡:Kafka的消费者可以组成消费者组,通过消费者组,可以将负载均匀地分配到多个消费者上,从而避免单个消费者的性能瓶颈,提高整个Kafka集群的可用性。
- 故障检测和恢复: Kafka 会使用 Zookeeper 等组件协助监控和管理集群的状态。当检测到故障节点时,就会自动将不可用的节点从集群中排除。而等到故障节点恢复后,也会重新将节点加入到集群当中。 (Kafka 2.8 后面可以脱离了 zookeeper启动,用 kraft 协议了,但后面的版本仍然支持 zk,但是不推荐使用了。)
集群高可用性是 Kafka非常关键的设计之一。通过多项机制组合,使得 Kafka 可以成为处理关键业务数据的可信平台。
Kafka中的消费者偏移量是如何管理的?
在Kafka中,消费者偏移量是指消费者在处理消息过程中所处的位置。Kafka中的消费者偏移量由两部分组成:Topic和Partition。对于每个消费者组,Kafka都会为其维护在每个 Partition 上的偏移量,以便在处理消息时可以准确地跟踪进度。
消费者偏移量的管理可以通过以下方式进行:
- 手动提交偏移量:消费者可以通过调用commitSync或commitAsync方法手动提交偏移量到Kafka。手动提交偏移量的方式需要开发者在适当的时机调用提交方法,确保消费者处理完消息后再提交偏移量。这种方式对于灵活性和精确控制偏移量非常有用,但需要开发者自行考虑提交的时机和异常处理。
- 自动提交偏移量:消费者可以配置为在后台自动提交偏移量。这意味着消费者会定期自动将已经处理的消息的偏移量提交给Kafka,而不需要开发者手动处理。通过配置参数enable.auto.commit为true,以及设置auto.commit.interval.ms参数来控制自动提交的频率。自动提交偏移量简化了管理,但可能会导致消息的重复处理或丢失,因此需要根据具体业务场景谨慎配置。
总之,Kafka 消费者的偏移量管理是确保消息传递的可靠性和一致性的重要部分。它允许消费者灵活地管理消息的消费进度,以满足不同的应用需求。无论您选择自动还是手动管理偏移量,都需要确保偏移量的正确提交,以避免消息的重复消费。
什么是“零拷贝”?有什么作用?
零拷贝是操作系统提供的一种优化 IO 操作的重要机制。通过零拷贝技术,操作系统可以极大的减少在一次 IO 操作中,数据从一个内存区域复制到另一个内存区域的次数,以及在此过程中对 CPU 的性能消耗。零拷贝技术可以极大的提高数据传输的效率,避免不必要的数据拷贝,从而降低系统负载。
零拷贝有两种实现方式,mmap文件映射和sendfile文件复制。
- mmap机制主要依赖于内存区域映射技术,可以减少一次 IO 操作中,内核态与用户态之间的数据传输,从而减少因为上下文切换而带来的 CPU 性能开销。mmap机制通常适合于对大量小文件的 IO 操作,Kafka 大量的运用 mmap 机制加速 Partition 日志文件的读写过程。
- sendfile主要依赖于 DMA 数据传输技术,采用一组单独的指令集来进行负责数据在内存不同区域之间的拷贝过程。这样就不再需要 CPU 来进行复制,从而减少 CPU 性能消耗,让 CPU 可以用于更重要的计算任务。sendfile通常适合于大文件的拷贝传输操作,Kafka 大量的运用 sendfile 机制,加速消息从 Partition 文件到网卡的传输过程。
总之,零拷贝是由操作系统提供的一种高效的文件读写技术,而 Kafka 则大量的运用了零拷贝技术,从而极大的提升了 Kafka 整体的工作性能。