MQ消息队列八股系列（3）| 小妖客栈

21、RocketMQ如何保证消息不丢失？

在 RocketMQ 中，消息的可靠性和一致性可以通过以下几个机制来保障：

1. 消息持久化

同步刷盘（SYNC_FLUSH）

描述：消息写入后立即同步刷盘，将消息持久化到磁盘。这种方式极大地提高了消息的可靠性，但会增加一定的延迟。
配置示例：

  flushDiskType=SYNC_FLUSH

异步刷盘（ASYNC_FLUSH）

描述：消息写入后进行异步刷盘，相比同步刷盘性能更高，但可靠性略低，因为在异步刷盘期间，如果发生故障可能会丢失未刷盘的消息。
配置示例：

  flushDiskType=ASYNC_FLUSH

2. 主从复制

RocketMQ 支持主从复制，通过将消息从主节点复制到从节点来提高消息的可靠性。

同步复制（SYNC_MASTER）

描述：消息写入主节点后，立即同步复制到从节点，确保消息在主从节点中都存在。同步复制提高了消息的可靠性，但会增加写入的延迟。
配置示例：

  brokerRole=SYNC_MASTER

3. 消息确认机制

RocketMQ 提供了消费者确认机制，确保消息被成功处理。

消费者确认

描述：消费者处理完消息后，需要向 RocketMQ 发送确认（ack）。如果消费者未能确认消息（例如，消费者崩溃），RocketMQ 会将消息重新投递给其他消费者进行处理。
示例代码：

  consumer.registerMessageListener(new MessageListenerConcurrently() {
      @Override
      public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
          // 消息处理逻辑
          return ConsumeConcurrentlyStatus.CONSUME_SUCCESS; // 处理成功
      }
  });

4. 分布式事务

RocketMQ 支持分布式事务，确保消息和业务操作的一致性。在分布式事务场景中，可以通过事务消息机制，确保消息与业务操作同步进行，不丢失。

示例代码：

  TransactionMQProducer producer = new TransactionMQProducer("producer_group");
  producer.start();

  Message msg = new Message("topic", "tag", "key", "body".getBytes());
  SendResult sendResult = producer.sendMessageInTransaction(msg, new LocalTransactionExecuter() {
      @Override
      public LocalTransactionState executeLocalTransactionBranch(Message msg, Object arg) {
          // 执行本地事务逻辑
          return LocalTransactionState.COMMIT_MESSAGE; // 提交事务
      }
  }, null);

5. 重试机制

RocketMQ 提供了消息重试机制，确保消息在处理失败时不会丢失。

生产者重试

描述：当生产者在消息发送失败时，可以通过配置重试机制进行多次尝试，确保消息成功发送。
配置示例：

  producer.setRetryTimesWhenSendFailed(3); // 设置重试次数

消费者重试

描述：当消费者在处理消息失败时，RocketMQ 会自动重试，直到消息被成功处理或达到最大重试次数。
配置示例：

  consumer.setMaxReconsumeTimes(5); // 设置最大重试次数

  consumer.registerMessageListener(new MessageListenerOrderly() {
      @Override
      public ConsumeOrderlyStatus consumeMessage(List<MessageExt> msgs, ConsumeOrderlyContext context) {
          // 顺序处理消息逻辑
          return ConsumeOrderlyStatus.SUCCESS; // 处理成功
      }
  });

通过以上机制，RocketMQ 能够确保消息的可靠性和一致性，适用于高可用性和分布式环境中的消息传递和处理。

22、RocketMQ如何做负载均衡？

在 RocketMQ 中，负载均衡是确保消费者能够高效、均匀地处理消息的关键机制。以下是关于 RocketMQ 负载均衡的详细说明：

1. 消息队列的分配

RocketMQ 的主题（Topic）被划分为多个消息队列（Message Queue），每个消息队列是一个独立的消息存储单元。消费者组中的每个消费者实例（Consumer Instance）会消费一个或多个消息队列。

2. 消费者组（Consumer Group）

消费者组是实现负载均衡的基础单位。一个消费者组中的多个消费者实例共同消费一个主题的消息队列。通过在消费者组内部进行消息队列的分配，RocketMQ 实现了负载均衡。

3. 负载均衡策略

RocketMQ 提供了多种负载均衡策略，常用的有以下几种：

3.1 平均分配（AllocateMessageQueueAveragely）

描述：这是 RocketMQ 的默认负载均衡策略。该策略将消息队列均匀地分配给消费者组中的每个消费者实例。
示例：假设有一个主题 TopicA，包含 4 个消息队列 Q1、Q2、Q3、Q4，而消费者组 GroupA 中有 2 个消费者实例 Consumer1 和 Consumer2。使用平均分配策略时，Consumer1 可能会分配到 Q1 和 Q2，而 Consumer2 分配到 Q3 和 Q4。

3.2 按环形分配（AllocateMessageQueueByCircle）

描述：该策略将消息队列按顺序循环分配给消费者实例。
示例：假设有 3 个消息队列 Q1、Q2、Q3，消费者组中有 2 个消费者实例 Consumer1 和 Consumer2。使用按环形分配策略时，Consumer1 可能会分配到 Q1 和 Q3，而 Consumer2 分配到 Q2。

3.3 自定义分配策略

描述：RocketMQ 允许用户实现自定义的负载均衡策略。用户可以通过实现 AllocateMessageQueueStrategy 接口，定义自己的消息队列分配逻辑。

4. 负载均衡过程

负载均衡过程通常在以下几种情况下触发：

消费者实例启动：当新的消费者实例加入消费者组时，RocketMQ 会重新分配消息队列。
消费者实例停止：当消费者实例停止或崩溃时，RocketMQ 会重新分配该实例负责的消息队列给其他存活的消费者实例。
定时任务：RocketMQ 内部有定时任务定期检查和调整消息队列的分配情况，确保负载均衡的持续有效。

5. 实现机制

RocketMQ 的负载均衡机制主要由以下几个组件实现：

RebalanceService：负责定期触发负载均衡过程。
AllocateMessageQueueStrategy：定义消息队列的分配策略。
ConsumerGroupInfo：维护消费者组的信息，包括消费者实例和消息队列的分配情况。

6. 负载均衡的注意事项

消费者实例的数量变化：当消费者实例的数量发生变化时，负载均衡过程会重新分配消息队列，这可能会导致短暂的消息消费停顿。
消息队列的数量变化：当主题的消息队列数量发生变化时，也会触发负载均衡过程。
消费进度管理：在负载均衡过程中，消费者实例需要妥善管理消费进度，确保消息不丢失或重复消费。

通过这些机制，RocketMQ 能够在分布式环境中高效、可靠地处理大量消息，同时保持消费者组内部的负载均衡。

23、RocketMQ在分布式事务下的底层原理？

RocketMQ 提供了分布式事务支持，通过两阶段提交（Two-Phase Commit）协议来确保消息传递的可靠性和一致性。以下是关于 RocketMQ 分布式事务的详细说明：

1. 事务消息发送流程

分布式事务消息的发送过程分为三个阶段：准备阶段、提交/回滚阶段和事务状态检查阶段。

1.1 准备阶段（Prepare Phase）

在准备阶段，消息生产者发送一条预备消息（Prepare Message）到 RocketMQ Broker。预备消息会被持久化到 Broker，但不会被消费者消费。这一阶段的目的是确保消息在正式提交之前不会被消费者看到。

示例代码：

TransactionMQProducer producer = new TransactionMQProducer("ProducerGroupName");
producer.setNamesrvAddr("localhost:9876");
producer.start();

// 定义事务监听器
TransactionListener transactionListener = new TransactionListenerImpl();
producer.setTransactionListener(transactionListener);

Message msg = new Message("TopicTest", "TagA", "Hello RocketMQ".getBytes());
TransactionSendResult result = producer.sendMessageInTransaction(msg, null);

1.2 提交/回滚阶段（Commit/Rollback Phase）

在本地事务执行完成后，消息生产者会根据本地事务的执行结果决定提交或回滚消息。提交消息会使得消息对消费者可见，而回滚消息则会删除预备消息。

事务监听器实现示例：

public class TransactionListenerImpl implements TransactionListener {
    @Override
    public LocalTransactionState executeLocalTransaction(Message msg, Object arg) {
        // 执行本地事务逻辑
        boolean success = executeLocalBusinessLogic();
        if (success) {
            return LocalTransactionState.COMMIT_MESSAGE; // 提交事务
        } else {
            return LocalTransactionState.ROLLBACK_MESSAGE; // 回滚事务
        }
    }

    @Override
    public LocalTransactionState checkLocalTransaction(MessageExt msg) {
        // 检查本地事务状态
        boolean success = checkLocalTransactionStatus();
        if (success) {
            return LocalTransactionState.COMMIT_MESSAGE; // 提交事务
        } else {
            return LocalTransactionState.ROLLBACK_MESSAGE; // 回滚事务
        }
    }
}

1.3 事务状态检查阶段（Transaction Status Check Phase）

如果在预备消息发送之后，由于网络或其他原因导致生产者未能及时提交或回滚事务，Broker 会定期向生产者询问事务的状态。生产者需要实现 checkLocalTransaction 方法来返回事务的实际状态。

2. 事务消息的存储和状态管理

RocketMQ 在 Broker 端会持久化预备消息，并在消息的元数据中记录其状态（准备中、已提交、已回滚）。当生产者提交或回滚事务时，Broker 会更新消息的状态。这种机制确保了消息的状态始终与事务的执行情况保持一致。

3. 消费者处理事务消息

消费者在消费消息时不会区分事务消息和普通消息。只有在事务消息被提交之后，消费者才能消费到这些消息。这种设计确保了消费者只能处理那些已成功提交的事务，从而避免了消费未完成事务的风险。

4. 事务消息的可靠性保证

消息持久化：RocketMQ 对预备消息进行持久化存储，确保消息不会丢失。
事务状态检查：通过事务状态检查机制，确保事务最终的一致性。
重试机制：在事务消息的各个阶段都有重试机制，确保消息的可靠传递和处理。

5. 事务消息的示例代码

完整的事务消息发送和处理代码示例：

public class TransactionProducer {
    public static void main(String[] args) throws MQClientException {
        TransactionMQProducer producer = new TransactionMQProducer("ProducerGroupName");
        producer.setNamesrvAddr("localhost:9876");
        producer.setTransactionListener(new TransactionListenerImpl());
        producer.start();

        Message msg = new Message("TopicTest", "TagA", "Hello RocketMQ".getBytes());
        try {
            TransactionSendResult result = producer.sendMessageInTransaction(msg, null);
            System.out.printf("%s%n", result);
        } catch (MQClientException e) {
            e.printStackTrace();
        }

        // 保持生产者运行，模拟执行本地事务
        Runtime.getRuntime().addShutdownHook(new Thread(() -> producer.shutdown()));
    }
}

class TransactionListenerImpl implements TransactionListener {
    @Override
    public LocalTransactionState executeLocalTransaction(Message msg, Object arg) {
        // 执行本地事务逻辑
        boolean success = executeLocalBusinessLogic();
        return success ? LocalTransactionState.COMMIT_MESSAGE : LocalTransactionState.ROLLBACK_MESSAGE;
    }

    @Override
    public LocalTransactionState checkLocalTransaction(MessageExt msg) {
        // 检查本地事务状态
        boolean success = checkLocalTransactionStatus();
        return success ? LocalTransactionState.COMMIT_MESSAGE : LocalTransactionState.ROLLBACK_MESSAGE;
    }

    private boolean executeLocalBusinessLogic() {
        // 模拟本地事务逻辑
        return true; // 事务成功
    }

    private boolean checkLocalTransactionStatus() {
        // 模拟检查本地事务状态
        return true; // 事务成功
    }
}

结论

RocketMQ 的分布式事务机制通过两阶段提交协议，确保消息的可靠性和一致性。通过预备消息、提交/回滚消息和事务状态检查机制，RocketMQ 能够在分布式环境中提供强有力的事务保障。

24、为什么不应该对所有的message 都使用持久化机制？

消息持久化是提高系统可靠性的重要手段，但也带来了性能开销、资源消耗、复杂性和维护成本等方面的挑战。以下是对这些方面的具体分析：

1. 性能开销

持久化操作涉及磁盘I/O，而磁盘I/O的速度远低于内存操作，因此频繁的持久化操作会导致系统性能下降。

写入延迟：每次消息持久化都需要写入磁盘，这增加了消息发送的延迟。
吞吐量降低：高频率的磁盘I/O会限制系统的整体吞吐量，影响系统的处理效率。

2. 资源消耗

持久化消息需要占用磁盘存储空间，尤其在高频率、大量消息传输的场景下，存储成本可能会显著增加。

磁盘空间占用：大量持久化消息会迅速消耗磁盘空间，可能需要频繁扩展存储容量。
存储成本增加：为满足高频消息传输的需求，可能需要投资更多的存储设备或选择更高容量的存储解决方案。

3. 复杂性和维护成本

持久化机制的引入增加了系统的复杂性，带来了额外的维护和管理需求。

管理复杂性：需要定期监控和清理过期的持久化消息，以确保磁盘空间的可用性和系统的稳定性。
维护成本：系统需要投入更多的人力和资源来维护持久化机制，例如磁盘监控、数据清理和故障处理等。

4. 应用场景需求

持久化并非适用于所有应用场景，尤其是那些对实时性要求高但对消息可靠性要求不高的场景。

实时性要求高的场景：例如实时数据流处理、临时通知等场景，消息持久化可能会导致不必要的延迟，反而影响实时性。
可靠性要求低的场景：例如缓存更新通知、统计数据等场景，丢失少量消息对系统整体影响不大，持久化可能显得不必要。

结论

虽然消息持久化可以显著提高系统的可靠性，但并非适用于所有场景。为了在保证系统可靠性的同时，最大化系统性能和资源利用效率，应该根据具体业务需求和系统性能要求合理选择消息持久化策略。

25、如果设计一个mq，你要考虑哪些？

这段内容详细描述了消息队列系统的架构设计和实现要点，涵盖了分布式架构、数据存储、消息传输、可靠性、安全性、性能优化等方面。以下是对该内容的总结与分析：

1. 架构设计

分布式架构：通过分区（Partition）和副本（Replication）机制，保证系统的高可用性和可靠性。每个分区独立进行读写操作，副本则保证数据的一致性和容错性。
集群管理：通过Broker和Zookeeper/Consul等组件来实现集群的管理和协调，确保分布式系统的稳定性。

2. 数据存储

日志形式存储：消息按顺序写入磁盘，保证数据持久化，同时可以通过建立索引文件提高查询效率。
内存缓存：对于热点数据的访问，可以将数据缓存到内存中，以提高读取速度，并通过内存队列快速处理消息的读写。

3. 消息模型

点对点模型（P2P）：每条消息只能被一个消费者读取，适用于单一消费者处理的场景。
发布/订阅模型（Pub/Sub）：消息可以被多个消费者读取，适用于广播类型的消息传递场景。

4. 消息传输

传输协议：支持TCP、HTTP/HTTPS、gRPC等多种传输协议，适应不同的网络环境和系统需求。
消息格式：支持JSON、Protobuf、Avro等不同格式，针对不同场景选择合适的序列化方式。

5. 消息处理

推/拉模式：根据场景需求选择消息推送或拉取模式，以平衡延迟和吞吐量。
消息确认：提供自动和手动确认两种方式，确保消息处理的可靠性。

6. 消息可靠性

消息持久化：支持同步和异步刷盘模式，平衡消息的可靠性与系统性能。
消息重试和死信队列（DLQ）：为处理失败的消息提供重试机制和后续处理手段，确保系统的稳定性。

7. 性能优化

批量处理：通过批量发送和消费消息，提升系统的传输和处理效率。
消息压缩：在传输和存储过程中对消息进行压缩，减少带宽占用和存储空间消耗。

8. 安全性

认证和授权：通过用户名/密码、OAuth等方式进行认证，并使用RBAC实现细粒度的权限控制。
数据加密：通过TLS/SSL加密传输和存储加密，确保数据的安全性。

9. 监控和运维

日志和指标：通过日志记录和性能指标监控系统运行状态，及时发现和处理问题。
报警和自动恢复：设置报警规则和自动故障恢复机制，确保系统的高可用性。

10. 可扩展性

水平扩展：通过增加节点和分片机制提升系统的处理能力。
垂直扩展：通过升级硬件提高单节点的性能。

总结

该架构设计全面考虑了消息队列系统的各个方面，从分布式架构到性能优化、安全性和监控运维，具有良好的可扩展性和可靠性，适用于各种复杂的分布式系统场景。

26、Kafka中的ISR、AR是什么？

ISR（In-Sync Replicas）和AR（Assigned Replicas）是与Kafka分区副本管理密切相关的两个概念。它们在确保Kafka集群的高可用性和数据一致性方面起着重要作用。以下是对这两个概念的详细解释：

1. ISR（In-Sync Replicas）

定义：ISR是指与领导者副本保持同步的一组副本。ISR集合中的副本与领导者副本的数据完全一致。
成员：ISR集合中的所有副本都是同步副本，即它们的数据与领导者副本保持一致。
同步条件：一个副本要进入ISR集合，必须满足以下条件：
在指定时间内（由参数replica.lag.time.max.ms控制），没有落后于领导者副本超过一定的滞后量（由参数replica.lag.max.messages控制）。
作用：ISR集合用于确保数据的高可用性和一致性。如果领导者副本发生故障，ISR集合中的一个副本会被选为新的领导者。

2. AR（Assigned Replicas）

定义：AR是指为一个分区分配的所有副本。它包括领导者副本和所有跟随者副本。
成员：AR集合中的副本包括领导者副本和所有跟随者副本，无论它们是否与领导者副本同步。
作用：AR集合用于描述一个分区的所有副本，包括那些可能暂时不同步的副本。

3. ISR与AR的关系与区别

包含关系：ISR是AR的子集。也就是说，ISR集合中的副本都是AR集合中的一部分，但AR集合中的副本不一定都在ISR集合中。
同步状态：ISR中的副本是与领导者副本同步的，而AR中的副本可能包括不同步的副本。

4. 示例说明

假设一个Kafka分区有三个副本，编号为0、1和2。副本0是领导者副本，副本1和2是跟随者副本。

全同步情况：如果副本1和2都与副本0同步，那么ISR集合为{0, 1, 2}。
部分不同步情况：如果副本2由于某种原因落后于副本0超过了允许的滞后时间或滞后消息数，副本2将从ISR集合中移除，此时ISR集合为{0, 1}。

5. 配置参数

replica.lag.time.max.ms：跟随者副本允许落后于领导者副本的最大时间，超过这个时间将被认为不同步。
replica.lag.max.messages：跟随者副本允许落后于领导者副本的最大消息数，超过这个数量将被认为不同步。

总结

ISR（In-Sync Replicas）：与领导者副本保持同步的一组副本，确保数据的高可用性和一致性。
AR（Assigned Replicas）：为一个分区分配的所有副本，包括领导者副本和所有跟随者副本。

27、Kafka中的ISR集合是什么？

ISR（In-Sync Replicas）集合在Kafka集群中起到了确保高可用性和数据一致性的关键作用。以下是对ISR集合的详细解释：

1. ISR 集合定义

ISR集合是指一个Kafka分区中所有与领导者副本保持同步的副本集合。这些副本的数据与领导者副本一致，并且能够在领导者副本发生故障时迅速接替成为新的领导者。

2. 详细说明

2.1 同步副本

领导者副本（Leader Replica）：每个分区都有一个领导者副本，负责处理所有的读写请求。
跟随者副本（Follower Replicas）：除了领导者副本，分区还拥有多个跟随者副本，它们从领导者副本复制数据，确保数据的一致性。

2.2 同步条件

时间滞后控制：一个副本要被认为是同步的，它必须在规定的时间内（由参数replica.lag.time.max.ms控制）不落后于领导者副本超过允许的时间。
消息滞后控制：通过replica.lag.max.messages参数，Kafka还可以控制副本滞后消息的最大数量。

2.3 ISR 集合的动态性

ISR集合是动态变化的：

当一个跟随者副本与领导者副本保持同步时，它会被加入到ISR集合中。
如果一个跟随者副本的滞后时间或滞后消息数超过了允许的阈值，它将被从ISR集合中移除。

3. 作用

3.1 高可用性

如果领导者副本发生崩溃，ISR集合中的一个副本会被选为新的领导者，确保分区的可用性和服务的连续性。

3.2 数据一致性

只有在消息被写入到ISR集合中的所有副本后，Kafka才会确认消息的写入成功。这一机制确保了数据的一致性。

4. 配置参数

replica.lag.time.max.ms：控制一个跟随者副本落后于领导者副本的最大时间，超过这个时间将被认为不同步。
replica.lag.max.messages：控制一个跟随者副本落后于领导者副本的最大消息数，超过这个数量将被认为不同步。

5. 示例

假设一个Kafka分区有三个副本，分别为0、1和2。副本0是领导者副本，副本1和2是跟随者副本。

全同步情况：如果副本1和2都与副本0同步，那么ISR集合为{0, 1, 2}。
部分不同步情况：如果副本2落后于副本0超过了允许的滞后时间或滞后消息数，副本2将从ISR集合中移除，此时ISR集合为{0, 1}。

6. 总结

ISR 集合：与领导者副本保持同步的一组副本，确保数据的高可用性和一致性。
动态变化：ISR集合会根据副本的同步状态动态变化，确保只有与领导者副本同步的副本在集合中。
高可用性和一致性：通过ISR集合，Kafka能够在领导者副本故障时快速选出新的领导者，并确保消息写入的可靠性。

28、kafka中的 zookeeper 是干嘛的？

ZooKeeper 在 Kafka 集群中扮演了至关重要的角色，主要负责集群的管理和协调工作，确保 Kafka 系统的稳定性和高可用性。以下是 ZooKeeper 在 Kafka 中的具体作用：

1. 集群元数据管理

Broker 信息：ZooKeeper 存储所有 Kafka Broker 的信息，包括它们的地址、端口等，这些信息用于协调 Broker 之间的通信。
主题和分区信息：ZooKeeper 保存所有主题及其分区的信息，包括每个分区的领导者副本和 ISR 集合，从而确保消息能够正确地路由到相应的分区。
消费者组信息：ZooKeeper 存储消费者组的偏移量信息和成员信息，帮助协调消息的消费进度。

2. 领导者选举

选举流程：在 Kafka 中，每个分区都有一个领导者副本，负责处理读写请求。当领导者副本发生故障时，ZooKeeper 会立即协调选举一个新的领导者副本，以确保分区的正常运行。

3. 配置管理

动态配置：ZooKeeper 存储 Kafka 集群的配置数据，并允许这些配置在运行时动态更新。例如，管理员可以通过 ZooKeeper 添加或删除主题，调整分区数量等。

4. 分布式协调

集群协调：Kafka 依赖 ZooKeeper 来协调集群中多个 Broker 之间的交互。当有新的 Broker 加入或现有 Broker 离开集群时，ZooKeeper 通知其他 Broker 进行相应的调整，以保持集群的一致性。

5. 偏移量存储

早期版本：在 Kafka 0.9 之前的版本中，消费者的偏移量信息是存储在 ZooKeeper 中的。这些偏移量信息用于追踪每个消费者读取的最新消息位置。
迁移后的改进：从 Kafka 0.9 开始，偏移量存储被迁移到了 Kafka 自身的特殊主题（_consumer_offsets）中，这种做法提升了性能和可靠性。

6. 监控和健康检查

集群健康监控：ZooKeeper 用于监控 Kafka 集群的健康状况。它能够检测到 Broker 的加入和退出，并触发相应的处理机制，确保集群的稳定运行。

工作流程示例

Broker 启动：Kafka Broker 启动时，会向 ZooKeeper 注册自己，并从中获取集群的配置信息。
主题创建：当创建一个新主题时，Kafka 将主题的配置信息（如分区数、副本数等）写入 ZooKeeper，以便集群其他组件使用。
领导者选举：如果一个分区的领导者副本发生故障，ZooKeeper 立即协调选举一个新的领导者副本，并更新集群元数据。

总结

ZooKeeper 在 Kafka 中的作用包括：

元数据管理：负责存储和管理 Kafka 集群的所有元数据。
领导者选举：确保每个分区在领导者副本故障时能够迅速选出新的领导者。
配置管理：支持 Kafka 配置的动态更新。
分布式协调：保持 Kafka 集群中多个 Broker 之间的协调一致。
监控和健康检查：监控 Kafka 集群的健康状态，提供故障检测和处理能力。

通过上述功能，ZooKeeper 确保了 Kafka 集群的高可用性、数据一致性和运行稳定性。

29、kafka的follower如何与leader同步数据？

Kafka 中的同步机制通过拉取模式（Pull-based Replication）实现，主要依赖 Follower 副本向 Leader 副本定期发送 Fetch 请求来保持数据同步。以下是详细的同步机制和流程：

同步机制

拉取数据：

Follower 副本通过周期性地向Leader 副本发送拉取请求（Fetch Request）来获取新的数据。这种方式称为拉取模式。

Fetch 请求：

Follower 副本会定期向Leader 副本发送 Fetch 请求，内容包括当前 Follower 副本已复制成功的最后一个消息的位置（偏移量）。
Leader 副本根据这个偏移量返回从该位置开始的新数据。

数据传输：

Leader 副本将从指定偏移量开始的消息数据打包成响应（Fetch Response），并返回给Follower 副本。
Follower 副本接收到这些消息后，将其追加到本地日志中。

确认和更新偏移量：

Follower 副本在成功写入本地日志后，会更新自己的偏移量，并在下一次 Fetch 请求中发送更新后的偏移量给Leader 副本，以便获取更多新数据。

详细步骤

初始同步：

当一个新的 Follower 副本加入到 ISR 集合中时，它首先需要进行初始同步。此时，Follower 副本会从 Leader 副本获取当前最新的日志，并将未同步的数据复制到本地。

持续同步：

完成初始同步后，Follower 副本会定期发送 Fetch 请求，以获取新的数据并保持与 Leader 副本的同步。

处理滞后：

如果 Follower 副本的复制滞后超过了配置的阈值（如 replica.lag.time.max.ms 或 replica.lag.max.messages），它将被从 ISR 集合中移除，直到重新完成同步。

配置参数

replica.fetch.max.bytes：Follower 副本在一次 Fetch 请求中能够拉取的最大数据量。
replica.fetch.wait.max.ms：Follower 副本在等待 Leader 副本返回 Fetch 响应时的最大等待时间。
replica.lag.time.max.ms：Follower 副本允许的最大滞后时间，超过这个时间将被认为不同步。
replica.lag.max.messages：Follower 副本允许的最大滞后消息数，超过这个数量将被认为不同步。

示例流程

假设有一个 Kafka 分区有三个副本：Leader 副本（副本 0）和两个 Follower 副本（副本 1 和副本 2）。

初始同步：

副本 1 和副本 2 启动，并向 Leader 副本 0 发送 Fetch 请求，获取当前最新的日志数据。
Leader 副本 0 返回从起始偏移量开始的日志数据，副本 1 和副本 2 将其写入本地日志。

持续同步：

副本 1 和副本 2 定期向 Leader 副本 0 发送 Fetch 请求，包含当前的偏移量。
Leader 副本 0 返回从该偏移量开始的新数据，副本 1 和副本 2 将其追加到本地日志。

处理滞后：

如果副本 1 或副本 2 的同步滞后超过配置的阈值，它们将被从 ISR 集合中移除，直到重新完成同步。

总结

Kafka 的 Follower 副本与 Leader 副本的同步通过拉取模式（Pull-based Replication）实现。Follower 副本定期向 Leader 副本发送 Fetch 请求，获取新的数据并将其追加到本地日志中，从而保持数据的一致性和同步性。

30、kafka为什么会出现重复消费？

Kafka 中的重复消费问题主要源于以下几个方面，并可以通过相应的策略来减少这种情况的发生。

重复消费的原因

消费者重启或崩溃：

当消费者重启或崩溃时，可能会重新读取某些消息，导致重复消费。这通常发生在消息处理完成但偏移量尚未提交的情况下。由于 Kafka 消费是基于偏移量的，如果偏移量未提交，消费者在重启后会从上次提交的偏移量开始重新消费，从而导致重复消费。

消费者组再平衡：

Kafka 使用消费者组来实现消息的负载均衡和容错性。当消费者组中的消费者数量发生变化（例如新增或移除消费者），Kafka 会触发再平衡操作。在再平衡期间，分区的所有权可能会转移到不同的消费者，这可能导致某些消息被重新处理，从而出现重复消费。

生产者重试机制：

当生产者发送消息失败时，可能会进行重试。如果生产者在重试过程中成功发送了消息，但消费者在处理过程中未能正确提交偏移量，则这些消息可能会被重新消费，导致重复消费。

消费者提交偏移量失败：

消费者在处理消息后需要提交偏移量，以告知 Kafka 已成功消费该消息。如果提交偏移量的操作失败（例如网络问题或 Kafka 集群故障），则消费者可能会在下次启动时重新消费这些消息，导致重复消费。

异步处理和提交偏移量：

在某些情况下，消费者可能会异步处理消息，并在处理完成后提交偏移量。如果在处理过程中发生故障，偏移量未能正确提交，则这些消息可能会被重新消费，导致重复消费。

分区副本切换：

在 Kafka 集群中，如果 Leader 副本发生故障，Kafka 会将 Leader 角色切换到其他 Follower 副本。在切换过程中，可能会导致某些消息的偏移量信息不一致，导致消费者重复消费这些消息。

减少重复消费的方法

使用幂等性生产者：

Kafka 提供了幂等性生产者（Idempotent Producer），确保每条消息只会被写入一次，即使发生重试。这可以减少由于生产者重试导致的重复消费。

使用事务性生产者：

Kafka 支持事务性生产者，允许生产者在一个事务中发送多条消息，并确保这些消息要么全部成功，要么全部失败。这样可以确保消息的原子性，减少重复消费。

手动提交偏移量：

使用手动提交偏移量（例如 enable.auto.commit=false），在消费者确认消息处理完成后再提交偏移量。这样可以确保只有在消息处理成功后才会提交偏移量，减少重复消费的可能性。

处理幂等性：

在消费者端实现幂等性处理，即使同一条消息被多次处理，也不会产生副作用。例如，可以使用唯一标识符（如消息的键）来确保每条消息只被处理一次。

优化消费者组再平衡：

尽量减少消费者组的频繁变动，优化再平衡策略。例如，可以使用 session.timeout.ms 和 max.poll.interval.ms 配置参数来调整再平衡的频率和超时时间。

总结

Kafka 出现重复消费的原因主要包括消费者重启或崩溃、消费者组再平衡、生产者重试机制、消费者提交偏移量失败、异步处理和分区副本切换。通过使用幂等性生产者、事务性生产者、手动提交偏移量、实现幂等性处理和优化消费者组再平衡等方法，可以有效减少重复消费的发生。

31、如何查看kafka消息积压？

查看 Kafka 消息积压情况的方法有很多，以下是常用的方法及其具体操作步骤：

1. 使用 Kafka 自带工具

kafka-consumer-groups.sh

Kafka 提供了 kafka-consumer-groups.sh 命令行工具来查看消费者组的消费情况。可以用来显示消费者组的当前偏移量、日志末尾偏移量和消息积压情况。

示例命令：

bin/kafka-consumer-groups.sh --bootstrap-server <broker_address> --describe --group <consumer_group>

输出说明：

CURRENT-OFFSET：消费者组当前的偏移量。
LOG-END-OFFSET：日志的末尾偏移量。
LAG：消息积压的数量，即 LOG-END-OFFSET - CURRENT-OFFSET。

2. 使用 Kafka Manager

Kafka Manager 是由 Yahoo 开源的工具，可以用来管理和监控 Kafka 集群。它提供了图形用户界面，可以直观地查看消费者组的消费情况和消息积压。

操作步骤：

安装并配置 Kafka Manager。
访问 Kafka Manager 的 Web 界面。
查看消费者组的详细信息，包括消息积压情况。

3. 使用 Kafka Exporter

Kafka Exporter 是一个 Prometheus 导出器，用于监控 Kafka 集群的状态。它将 Kafka 的监控数据导出到 Prometheus，可以通过 Grafana 等工具进行可视化。

操作步骤：

安装 Kafka Exporter。
配置 Kafka Exporter 以连接到 Kafka 集群。
将监控数据导出到 Prometheus。
使用 Grafana 等工具可视化消息积压情况。

4. 使用自定义脚本

可以编写自定义脚本，利用 Kafka 的 AdminClient API 来获取消息积压情况。以下是一个简单的 Java 示例：

示例代码：

import org.apache.kafka.clients.admin.AdminClient;
import org.apache.kafka.clients.admin.AdminClientConfig;
import org.apache.kafka.clients.admin.ListConsumerGroupOffsetsResult;
import org.apache.kafka.clients.admin.OffsetSpec;
import org.apache.kafka.clients.admin.TopicPartition;
import org.apache.kafka.clients.admin.OffsetAndMetadata;

import java.util.Collections;
import java.util.Map;
import java.util.Properties;
import java.util.concurrent.ExecutionException;

public class KafkaLagChecker {
    public static void main(String[] args) throws ExecutionException, InterruptedException {
        Properties props = new Properties();
        props.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");

        try (AdminClient adminClient = AdminClient.create(props)) {
            String groupId = "your-consumer-group";
            ListConsumerGroupOffsetsResult offsetsResult = adminClient.listConsumerGroupOffsets(groupId);

            Map<TopicPartition, OffsetAndMetadata> offsets = offsetsResult.partitionsToOffsetAndMetadata().get();
            Map<TopicPartition, Long> logEndOffsets = adminClient.listOffsets(Collections.singletonMap(
                new TopicPartition("your-topic", 0), OffsetSpec.latest()
            )).all().get();

            for (Map.Entry<TopicPartition, OffsetAndMetadata> entry : offsets.entrySet()) {
                TopicPartition tp = entry.getKey();
                long currentOffset = entry.getValue().offset();
                long logEndOffset = logEndOffsets.get(tp);
                long lag = logEndOffset - currentOffset;

                System.out.printf("Topic: %s, Partition: %d, Lag: %d%n", tp.topic(), tp.partition(), lag);
            }
        }
    }
}

总结

查看 Kafka 消息积压的常用方法包括：

使用 Kafka 自带的命令行工具（如 kafka-consumer-groups.sh）；
第三方监控工具（如 Kafka Manager、Kafka Exporter）；
编写自定义脚本，利用 Kafka 的 AdminClient API。

这些方法可以帮助你有效地监控 Kafka 消息积压情况，确保系统的正常运行。

32、kafka怎么修改偏移量？

要重置 Kafka 消费者组的偏移量，有多种方法可以选择。以下是常用的几种方法及其操作步骤：

1. 使用 `kafka-consumer-groups.sh` 工具

重置偏移量到最新位置

bin/kafka-consumer-groups.sh --bootstrap-server <broker_address> --group <consumer_group> --topic <topic> --reset-offsets --to-latest --execute

重置偏移量到最早位置

bin/kafka-consumer-groups.sh --bootstrap-server <broker_address> --group <consumer_group> --topic <topic> --reset-offsets --to-earliest --execute

重置偏移量到特定时间点

假设要将偏移量重置到 2023-07-07 12:00:00 的时间点：

bin/kafka-consumer-groups.sh --bootstrap-server <broker_address> --group <consumer_group> --topic <topic> --reset-offsets --to-datetime 2023-07-07T12:00:00 --execute

重置偏移量到特定偏移量

假设要将偏移量重置到 100：

bin/kafka-consumer-groups.sh --bootstrap-server <broker_address> --group <consumer_group> --topic <topic> --reset-offsets --offsets-for-times 100 --execute

重置偏移量相对于当前偏移量

假设要将偏移量前移 10 条消息：

bin/kafka-consumer-groups.sh --bootstrap-server <broker_address> --group <consumer_group> --topic <topic> --reset-offsets --shift-by -10 --execute

2. 使用 Kafka AdminClient API

示例代码：

import org.apache.kafka.clients.admin.AdminClient;
import org.apache.kafka.clients.admin.AdminClientConfig;
import org.apache.kafka.clients.admin.AlterConsumerGroupOffsetsResult;
import org.apache.kafka.clients.admin.OffsetAndMetadata;
import org.apache.kafka.common.TopicPartition;

import java.util.Collections;
import java.util.Map;
import java.util.Properties;
import java.util.concurrent.ExecutionException;

public class KafkaOffsetResetter {
    public static void main(String[] args) throws ExecutionException, InterruptedException {
        Properties props = new Properties();
        props.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");

        try (AdminClient adminClient = AdminClient.create(props)) {
            String groupId = "your-consumer-group";
            TopicPartition tp = new TopicPartition("your-topic", 0);
            OffsetAndMetadata offsetAndMetadata = new OffsetAndMetadata(100); // 设置偏移量
            Map<TopicPartition, OffsetAndMetadata> offsets = Collections.singletonMap(tp, offsetAndMetadata);
            AlterConsumerGroupOffsetsResult result = adminClient.alterConsumerGroupOffsets(groupId, offsets);
            result.all().get();
            System.out.println("Offset reset successfully.");
        }
    }
}

3. 使用 Kafka Streams API

可以使用 Kafka Streams API 的 cleanup() 方法来重置偏移量。这个方法会删除本地状态存储，并将偏移量重置到最早位置。

示例代码：

import org.apache.kafka.streams.KafkaStreams;
import org.apache.kafka.streams.StreamsConfig;

import java.util.Properties;

public class KafkaStreamsOffsetReset {
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put(StreamsConfig.APPLICATION_ID_CONFIG, "your-app-id");
        props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");

        KafkaStreams streams = new KafkaStreams(yourTopology(), props);
        streams.cleanUp(); // 清理本地状态并重置偏移量
    }
}

4. 使用 Kafka Connect

可以通过 Kafka Connect 的配置参数 consumer.override.auto.offset.reset 来控制消费者的偏移量重置策略。例如，将其设置为 earliest 或 latest。

示例配置：

{
  "name": "your-connector",
  "connector.class": "org.apache.kafka.connect.file.FileStreamSourceConnector",
  "tasks.max": "1",
  "file": "/path/to/your/file",
  "topic": "your-topic",
  "consumer.override.auto.offset.reset": "earliest"
}

总结

kafka-consumer-groups.sh 工具：适用于快速重置偏移量到特定位置、时间点或相对位置。
Kafka AdminClient API：适用于编程实现偏移量的修改。
Kafka Streams API：适用于重置本地状态存储和偏移量。
Kafka Connect：通过配置调整消费者的偏移量重置策略。

选择适合的方法根据实际需求进行操作。

33、Kafka消息丢失原因？

在 Kafka 环境中，消息丢失的常见原因及其预防措施如下：

1. 生产者问题

a. 未启用 ACK 机制

问题：如果生产者没有启用 acks 参数，或者将其设置为 acks=0，消息发送后不会等待任何确认，这可能导致消息在网络或服务器故障时丢失。
预防：将 acks 参数设置为 all 或 -1，确保消息被所有副本确认。

  Properties props = new Properties();
  props.put("acks", "all"); // 确保消息被所有副本确认

b. 重试机制配置不当

问题：生产者未启用重试机制，或者重试次数设置过少，可能导致消息在发送失败时不被重新发送。
预防：启用重试机制，并合理配置重试次数。

  props.put("retries", "3"); // 设置重试次数

2. 代理（Broker）问题

a. 副本不足

问题：如果 Kafka 集群中的某个主题副本数不足（例如，只有一个副本），当该副本所在的代理节点宕机时，消息可能会丢失。
预防：确保主题有足够的副本数，并配置 min.insync.replicas。

  bin/kafka-configs.sh --alter --entity-type topics --entity-name your_topic --add-config min.insync.replicas=2

b. 数据写入磁盘前代理宕机

问题：如果消息在写入磁盘前代理节点宕机，消息可能会丢失。
预防：确保代理配置中的 log.flush.interval.messages 和 log.flush.interval.ms 参数设置合理。

3. 消费者问题

a. 自动提交偏移量

问题：如果消费者启用了自动提交偏移量（enable.auto.commit=true），但在处理消息之前宕机或出错，偏移量已经提交，但消息未处理，导致消息丢失。
预防：禁用自动提交，并在处理消息后手动提交偏移量。

  props.put("enable.auto.commit", "false"); // 禁用自动提交

b. 消费者重平衡

问题：在消费者重平衡过程中，如果偏移量未正确提交，可能导致消息重复消费或丢失。
预防：在消费者重平衡期间，确保偏移量正确提交，或使用分区重新分配策略来减少重平衡带来的影响。

4. 网络问题

a. 网络分区

问题：在网络分区的情况下，生产者和消费者可能无法与 Kafka 集群通信，导致消息丢失或延迟。
预防：确保网络稳定，并设置适当的超时和重试机制来处理网络问题。

5. 配置问题

a. 主题配置不当

问题：主题的 retention.ms 或 retention.bytes 参数配置不当，可能导致消息在还未被消费前被删除。
预防：确保主题的保留策略合理，防止消息过早删除。

  bin/kafka-topics.sh --alter --topic your_topic --config retention.ms=604800000 # 设置保留时间为7天

b. 副本配置不当

问题：副本同步配置不当，如 min.insync.replicas 设置过低，可能导致在部分副本不可用时消息丢失。
预防：配置合适的副本同步设置。

  bin/kafka-configs.sh --alter --entity-type topics --entity-name your_topic --add-config min.insync.replicas=2

6. 磁盘问题

a. 磁盘故障

问题：磁盘故障可能导致数据丢失，尤其是在没有足够的副本时。
预防：确保每个主题有足够的副本，并定期检查磁盘健康状态。

7. 其他问题

a. 人为错误

问题：人为错误，如误操作删除主题、误配置参数等，也可能导致消息丢失。
预防：进行适当的权限控制，定期备份重要数据，并进行配置管理和审计。

总结

为了减少消息丢失的风险，建议：

启用并正确配置生产者的 ACK 机制和重试机制。
确保 Kafka 集群中有足够的副本，并设置合理的副本同步配置。
禁用自动提交偏移量，改为手动提交。
确保网络稳定，并设置适当的超时和重试机制。
合理配置主题的保留策略，并确保副本配置正确。
定期检查磁盘健康状态，并进行数据备份。
实施权限控制和配置管理，防止人为错误。

通过这些措施，可以有效降低消息丢失的风险。

34、Kafka、 ActiveMQ、 RabbitMQ、RocketMQ有什么优缺点？

你对各大消息队列系统（Kafka、ActiveMQ、RabbitMQ 和 RocketMQ）的优缺点总结得很全面。以下是每种系统的特点及其适用场景的详细说明：

Apache Kafka

优点：

高吞吐量：设计用于处理高吞吐量的实时数据流，每秒能处理数百万条消息。
持久化和可靠性：消息持久化到磁盘，副本机制保证高可用性和可靠性。
水平扩展：通过分区机制支持水平扩展，增加代理节点可以处理更多数据。
强大的社区支持：有活跃的开源社区和丰富的文档，支持多种生态系统集成，如 Kafka Streams 和 Kafka Connect。
灵活的消费模型：支持发布-订阅和队列模型，消费者可以根据需要读取消息。

缺点：

复杂的运维：部署和运维较为复杂，需要专门的运维人员来管理。
高延迟：相对于内存中消息传递的系统，磁盘操作可能带来一定的延迟。
功能单一：主要专注于高吞吐量和持久化，对消息路由和优先级队列等高级特性支持不如其他系统。

适用场景：适合需要处理大量实时数据流的高吞吐量场景，如大数据处理和实时分析。

ActiveMQ

优点：

成熟稳定：经过多年的发展和优化，具有较高的稳定性。
丰富的特性：支持多种消息传递模型（点对点、发布-订阅）、消息持久化、事务、消息优先级、延迟消息等。
多协议支持：支持多种协议（AMQP、MQTT、STOMP、OpenWire），灵活性高。
简单易用：相对容易部署和使用，适合中小型企业和应用。

缺点：

性能瓶颈：在高吞吐量和高并发场景下，性能较 Kafka 和 RocketMQ 略显不足。
扩展性有限：扩展性和水平扩展能力不如 Kafka 和 RocketMQ。

适用场景：适合需要丰富消息传递特性和多协议支持的中小型应用。

RabbitMQ

优点：

灵活的路由机制：提供复杂的消息路由机制，如交换器、绑定键、队列，支持多种消息传递和路由模式。
多协议支持：支持 AMQP、MQTT、STOMP 等多种协议，适用性广。
易于使用：提供丰富的客户端库和管理工具，易于上手。
可靠性高：支持消息持久化、确认机制、事务，保证消息的可靠传递。

缺点：

性能限制：在极高吞吐量和低延迟场景下，性能不如 Kafka 和 RocketMQ。
运维复杂度：在大规模集群中，运维和管理相对复杂。
内存消耗：处理大量消息时，内存消耗较高，需要合理配置和管理。

适用场景：适合需要复杂消息路由和灵活性的应用，尤其是在中等吞吐量和低延迟需求的场景。

Apache RocketMQ

优点：

高性能：设计用于高吞吐量和低延迟的消息传递，性能接近 Kafka。
强大的消息路由：支持复杂的消息路由和过滤机制。
分布式事务：支持分布式事务，适合需要严格事务保证的场景。
扩展性强：支持水平扩展，能够轻松扩展集群规模。
可靠性高：支持消息持久化、副本机制，保证消息的高可用性和可靠性。

缺点：

社区和生态：社区和生态系统相对较小。
运维复杂度：部署和运维相对复杂，需要专业知识和经验。
文档和支持：文档和社区支持可能不如更成熟的系统丰富。

适用场景：适合高性能、低延迟和分布式事务需求的场景，尤其是在需要高扩展性和高可靠性的情况下。

总结

Kafka：高吞吐量和实时数据流处理，适合大规模数据持久化。
ActiveMQ：丰富特性和多协议支持，适合中小型应用。
RabbitMQ：复杂消息路由和灵活性高，适合中等吞吐量和低延迟场景。
RocketMQ：高性能、低延迟和分布式事务，适合高扩展性和高可靠性需求。

根据具体的应用需求和场景选择合适的消息队列系统可以更好地满足性能和功能要求。

此处内容已隐藏，请评论后刷新页面查看.

版权声明 1 本站名称： 小妖客栈
2 本站网址：www.wangkay.top
3 本网站的文章部分内容可能来源于网络，仅供大家学习与参考，如有侵权，请联系站长进行删除处理。
4 本站一切资源不代表本站立场，并不代表本站赞同其观点和对其真实性负责。
5 本站一律禁止以任何方式发布或转载任何违法的相关信息，访客发现请向站长举报。
6 本站附件资源、教程等内容如因时效原因失效或不可用，请评论区留言或联系站长及时更新。

THE END

八股文中间件
# java # 中间件