问题从一次Kafka宕机说起。笔者是一家金融科技公司，但是公司并没有使用金融支付领域比较流行的RabbitMQ，而是使用了Kafka，一开始是为日志处理而设计的，所以一直对Kafka的高可用很好奇实施和保证。从Kafka部署后，系统内部使用的Kafka一直稳定运行，没有出现不可用的情况。但是最近经常有系统测试人员反映Kafka消费者偶尔收不到消息。当他们登录到管理界面时，发现三个节点中有一个节点宕机挂掉了。但是按照高可用的概念，三节点两节点可用，怎么可能导致整个集群的消费者收不到消息呢？要解决这个问题，就要从Kafka的高可用实现入手。Kafka的多副本冗余设计无论是基于关系型数据库设计的传统系统，还是分布式系统如zookeeper、redis、Kafka、HDFS等，实现高可用的方式通常是采用冗余设计。解决节点宕机不可用问题。首先简单了解一下Kafka的几个概念：逻辑模型Broker（节点）：Kafka服务节点，简单来说一个Broker就是一个Kafka服务器和一个物理节点。主题（topic）：在Kafka中，消息以主题为单位进行分类。每个主题都有一个主题名称。生产者根据TopicName向特定的Topic发送消息，消费者也根据TopicName从对应的Topic发送消息。消费。Partition（分区）：Topic（主题）是消息分类的一个单位，但每个topic又可以细分为一个或多个Partition（分区），一个partition只能属于一个topic。主题和分区都是逻辑概念。比如消息1和消息2都发送到topic1，它们可能进入同一个分区，也可能进入不同的分区（所以同一个topic下不同分区包含的消息是不同的），然后发送给Broker分区对应的节点。偏移量（offset）：一个partition可以看做一个只能进出的队列（Kafka只保证一个partition中的消息是有序的）。消息将附加到此队列的末尾。每条消息进入分区后，都会有一个偏移量，它标识消息在分区中的位置，消费者通过偏移量来识别消息。就这么简单？是的，Kafka的高可用就是基于上面的多副本架构图实现的。当一个Broker挂了，不用担心，这个Broker上的Partition在其他Broker节点上还有副本。挂断电话的是领导，你说呢？然后就在Followers中选出一个Leader，生产者和消费者又可以和新的Leader一起愉快的玩耍了。这是高可用性。你可能还有疑问，多少份才够？如果Follower和Leader没有完全同步怎么办？节点宕机后的leader选举规则是什么？直截了当的结论：多少副本就足够了？副本越多，越能保证Kafka的高可用，但是副本越多，对网络和磁盘资源的消耗就越大，性能也会下降。一般来说，副本数为3，以保证高可用。在极端情况下，只需增加replication-factor参数即可。如果follower和lead不完全同步怎么办？Follower和Leader不是完全同步的，但也不是完全异步的，而是使用了ISR机制（In-SyncReplica）。每个Leader都会动态维护一个ISR列表，里面存放着与Leader基本同步的Follower。如果一个Follower因为网络、GC等原因没有向Leader发起拉取数据请求，此时Follower与Leader不同步，就会被踢出ISR列表。因此，ISR列表中的Follower是能跟上Leader的副本。节点宕机后的leader选举规则是什么？分布式选举规则有很多，比如Zookeeper的Zab、Raft、ViewstampedReplication、微软的PacificA等，Kafka的leader选举的思路很简单。根据我们上面提到的ISR列表，它会在关机后依次从所有副本中搜索。如果找到的副本在ISR列表中，它将被选为领导者。另外，还要保证之前的Leader已经退位，否则会出现脑裂的情况（有两个Leader）。如何保证？Kafka通过设置controller来保证只有一个Leader。Ack参数决定了可靠性的程度。另外这里补充一个面试Kafka高可用的知识点：request.required.asks参数。asks这个参数是生产者客户端的一个重要配置，可以在发送消息的时候设置。该参数具有三个可配置值：0、1、All。第一种是设置为0，即生产者发送消息后，我们不关心消息是死是活。如果不负责任，消息可能会丢失，可用性也会丢失。second设置为1，表示producer发送消息后，只要消息成功传递给Leader，其他Follower是否同步都无所谓。有一种情况，Leader刚收到消息，Follower还没来得及同步Broker就宕机了，但是producer已经认为消息发送成功了，所以这个时候消息就丢失了。将其设置为1是Kafka的默认配置。可以看出，Kafka默认的配置并不是那么高可用，而是在高可用和高吞吐量之间的权衡。三是设置为All（或-1），即producer发送消息后，不仅Leader要接收，ISR列表中的Follower也要同步，producer会发送任务消息成功。再想一想，Asks=All不就不会丢消息了吗？答案是不。当ISR列表中只剩下Leader时，Asks=All相当于Asks=1。这样的话，如果节点宕机了，数据能不丢吗？因此，只有在Asks=All并且ISR中有两个副本时，才能保证数据丢失。解决问题兜兜转转，了解了Kafka的高可用机制，最后又回到了我们最初的问题本身，为什么一个Kafka节点宕机后不可用？我在开发测试环境配置的Broker节点数为3个，Topic副本数为3个，Partitions个数为6个，Asks参数为1，当三个节点其中一个宕机时，会发生什么集群先做？没错，我们上面说了，当集群发现有Partition的Leader失效时，此时需要从ISR列表中重新选举Leader。ISR列表为空是不是不可用？不是，而是选择其中一个Partition的幸存副本作为领导者，但存在数据丢失的潜在风险。因此，只要设置Topic副本数与Brokers数量相同，Kafka的多副本冗余设计就可以保证高可用，宕机后不会不可用（但需要注意的是Kafka有一个保护策略，当超过一半的节点不可用时，Kafka停止）。那么仔细想想，Kafka上有没有副本数为1的topic？问题出在__consumer_offset。__consumer_offset是Kafka自动创建的topic，用于存储消费者消费的偏移量（offset）信息。默认的partition数量是50。而这个Topic，它的默认副本数量是1。如果所有Partition都存在于同一台机器上，那是很明显的单点故障！当把存放__consumer_offset的Partition的Broker给Kill时，会发现所有的consumer都停止消费了。如何解决这个问题呢？第一点是需要删除__consumer_offset。注意这个topic是kafka内置的topic，不能用命令删除。我通过删除日志来删除它。第二点，需要通过设置offsets.topic.replication.factor为3，将__consumer_offset的副本数改为3。通过__consumer_offset作为副本冗余，解决了某个节点宕机后消费者的消费问题。最后，我很困惑为什么__consumer_offset的Partition只存储在一个Broker上，而不是分布在各个Broker上。
