首页  思维导图  详情

Kafka权威指南

2018-12-11 11:10:51   0  举报





AI智能生成

《Kafka权威指南》读书笔记

kafka

消息队列

Kafka核心概念

作者其他创作

大纲/内容

安装Kafka

1.安装Zookeeper

Kafka内置了Zookeeper，但是独立安装一个完整的Zookeeper更好一些

2.安装Kafka Broker

下载地址: http://kafka.apache.org/downloads

多节点

config/server-1.properties:

broker.id=1

listeners=PLAINTEXT://:9093

log.dirs=/tmp/kafka-logs-1

config/server-2.properties:

broker.id=2

listeners=PLAINTEXT://:9094

log.dirs=/tmp/kafka-logs-2

> tar -xzf kafka_2.11-2.1.0.tgz

> cd kafka_2.11-2.1.0

> bin/zookeeper-server-start.sh config/zookeeper.properties
> bin/kafka-server-start.sh config/server.properties(单节点)

3.配置Broker

参考《配置篇》节点

4.测试

1.创建Kafka Topic

> bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic test

2.查看Topic列表

> bin/kafka-topics.sh --list --zookeeper localhost:2181

3.创建生产者，并写入消息

> bin/kafka-console-producer.sh --broker-list localhost:9092 --topic test

This is a message

This is another message

4.创建消费者，订阅消息

> bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test --from-beginning

This is a message

This is another message

5.更多参考

https://kafka.apache.org/quickstart

Kafka Consumer

理论篇

3.再均衡

1.概念

：分区的所有权从一个消费者转移到另一个消费者，这样的行为被称为再均衡

2.再均衡的优势

：再均衡给消费者群组带来了高可用性和伸缩性

3.再均衡的劣势：

3.1再均衡期间，消费者无法读取消息，会造成整个群组一小时间段的不可用

3.2当分区被重新分配给另一个消费者时，消费者当前读取的状态会丢失，它有可能还需要去刷新缓存，在它重新恢复状态之前会拖慢应用程序——如何安全的再均衡以及如何避免不必要的再均衡？

4.群组协调器：

消费者通过向被指派为群组协调器的broker发送心跳来维持它们和群组的从属关系以及它们对分区的所有权关系

5.心跳保活：

1.消费者只要以正常的时间间隔发送心跳，就被认为是活跃的——它还在读取分区里的消息

2.消费者会在轮询消息或提交偏移量时发送心跳

3.如果消费者停止发送心跳的时间足够长，会话就会过期，群组协调器就会认为它已经死亡，就会触发一次再均衡

4.在0.10.1版本里，Kafka社区引入了一个独立的心跳线程，可以在轮询消息的空档发送心跳

6.分配分区是怎样的一个过程

6.1当消费者要加入群组时，它会向群组协调器发送一个JoinGroup请求，第一个加入群组的消费者将成为群主

6.2它使用了一个实现了PartitionAssignor接口的类来决定哪些分区应该被分配给哪个消费者

6.3kafka内置了两种分配策略

6.3.1.Range，该策略会把主题的若干个连续的分区分配给消费者

6.3.2.RoundRobin，该策略把主题的所有分区逐个分配给消费者

6.4每个消费者只能看到自己的分配信息，只有群主知道群组里所有消费者的分配信息，这个过程会在每次再均衡时重复发生

2.消费时可能会出现的问题

1.消费者群组中多个消费者读取消息时，如果一个消费者被关闭或者发生崩溃时，它就离开群组，原本由它读取的分区将由组内其它消费者来读取

2.在主题发生变化时，比如管理员添加了新的分区，会发生分区重分配

1.消费者和消费者群组

消费者群组：消费者是消费者群组的一部分，也就是说会有一个或多个消费者共同读取一个主题；一个群组里的消费者订阅的是同一个主题，每个消费者接收主题一部分分区的消息

消费者的个数一定小于等于分区数

消费者的速度如果做一些高延迟的操作，比如把数据写入到DB或者HDFS或者进行比较耗时的计算，这时就会增加更多的消费者

群组之间相互独立

实战篇

创建Kafka消费者

订阅主题

轮询

提交和偏移量

定义

更新分区当前位置的操作叫做提交

如何提交？

消费者往一个叫做_consumer_offset的特殊主题发送消息，消息里包含每个分区的偏移量

消息为什么会重复？

如果提交的偏移量小于客户端处理的最后一个消息的偏移量，那么处于两个偏移量之间的消息就会被重复处理

消息为什么会丢失

如果提交的偏移量大于客户端处理的最后一个消息的偏移量，那么处于两个偏移量之间的消息将会丢失

提交的方式/种类

自动提交

最简单的方式就是自动提交，默认值5s，自动提交也是在轮询里进行的，消费者每次在进行轮询时会检查是否该提交偏移量了，如果是，那么就就会提交从上次轮询返回的偏移量

自动提交缺陷

如果我们使用默认的5s间隔，那么如果在最后一次提交之后的3s内发生再均衡，消费者从最后一次提交的偏移量位置开始读，这时候偏移量已经落后了3s，所以这3s内的消息将会被重复处理

异步提交

commitAsync()

异步提交可能带来消息重复

默认异步提交是不进行重试的，是因为在它收到响应的时候，可能有一个更大的偏移量已经提交成功

异步提交的重试

异步提交可以使用回调进行重试

我们可以使用一个单调递增的序列号来维护异步提交的顺序，在每次提交偏移量之后或者在回调里提交偏移量时递增序列号

异步提交重试需要注意维护提交顺序

在重试之前要先检查回调的序列号和即将的提交的偏移量是否相等，如果相等证明没有新的提交，那么可以进行安全的重试，否则应该停止重试

提交特定的偏移量

使用场景

如果我们poll()时一次返回了一大批数据，为了避免再均衡引起的重复提交，我们可能需要在批次中间进行提交

commitAsync() & commitSync()只会提交最后一个偏移量，而此时整个批次的消息还没处理完

如何提交特定的偏移量？

消费者API允许在使用CommitAsync()和commitSync()时传入指定的偏移量

风险

消费者可能不只读取一个分区，我们需要跟踪所有分区的偏移量——这是相对复杂的

同步和异步组合提交

使用场景

一般情况下，针对偶尔提交失败，不进行重试不会有太大问题，因为如果提交失败是因为临时问题导致，那么后续的提交总会有成功的，但是如果是发生在关闭你消费者或再均衡前的最后一次提交，就要确保提交成功

使用方法

先使用异步提交，这样速度更快，即使提交失败，可能下次也成功了，然后在finally中使用同步提交，这样就会一直重试，直到提交成功或者发生无法恢复的错误。

提交当前偏移量

大部分开发者通过控制偏移量提交时间来消除丢失消息的可能性，并在发生再均衡时减少重复消息的数量

消费者API提供了一种可以不基于时间间隔的提交偏移量的方式，

此时需要先将auto.commit.offset设置为false,然后由程序决定何时提交偏移量

使用commitSync()提交偏移量，最简单也最可靠，成功就马上返回，失败就抛出异常

只要没有发生不可恢复的错误，commitSync()方法会一直尝试直至提交成功，如果失败，我们也只能记录日志

手动提交的缺陷

在broker对提交请求做出回应之前，应用程序会一直阻塞，会限制程序的吞吐量

再均衡监听器

消费者在退出和分区再均衡之前会做一些清理操作

1.会在消费者失去对一个分区的所有权之前提交最后一个已处理记录的偏移量。

2.如果消费者准备了一个缓冲区用于处理偶发的事件，那么在失去分区所有权之前，需要处理在缓存区积累下来的记录

3.还可能需要关闭文件句柄、数据库连接等

在为消费者分配新分区或移除旧分区时，可以通过消费者API执行一些应用程序代码，

在调用subscribe()方法时传进去一个ConsumerRebalanceListener实例就可以了

ConsumerRebalanceListener的两个方法

public void OnpartitionRevoked(Collection<TopicPartition>)

该方法会在再均衡开始之前和消费者停止读取消息之后被调用，如果在这里提交偏移量，下一个接管分区的消费者就知道该从哪里开始读取了。

public void onPartitionAssigned(Collection<TopicPartition> partitions)

该方法会在重新分配分区之后和消费者开始读取消息之前被调用

从特定偏移量处开始处理记录

poll()是从最新的偏移量开始处理消息

从分区的起始位置开始读取消息

seekToBeginning(Collection<TopicPartition> tp)

从分区的末尾开始读取消息

seekToEnd(Collection<TopicPartition> tp)

从一个指定的位置开始读取

seek(TopicPartition tp, int offset)

退出

不需要担心消费者会在一个无限循环里

如果确定要退出循环，需要用另一个线程调用consumer.wakeup()

如果循环运行在主线程里，那么可以在ShutdownHook里调用该方法

consumer.wakeup()是消费者唯一一个可以从其他线程里安全调用的方法

调用consumer.wakeup()可以退出poll()，同时抛出WakeupException()异常，不需要处理这个异常，它只是退出循环的一种方式

在退出线程之前执行consumer.close()是很有必要的，它会执行所有还没提交的东西，并向群组协调器发送消息，告知自己要离开群组，接下来就会触发再均衡而不需要等待超时

序列化与反序列化

独立的消费者

如果只需要一个消费者从一个主题或者指定的分区读取数据，那么就不需要订阅主题，取而代之的是为自己分配分区

自己手动分配分区assign()

缺陷

如果主题增加了分区，消费者并不会得到通知，需要我们自己手动检查

Demo

子主题

配置篇

fetch.min.bytes

从服务器获取记录的最小字节数；broker收到消费者的数据请求时，如果可用数据量小于该配置，那么它会等到有足够的可用数据时才把它返回给消费者，这样可以降低消费者和broker的工作负载，在主题不活跃的时段就不需要来回处理消息

fetch.max.wait.ms

broker的等待时间，默认是500ms，如果没有足够的数据，就会使用该配置也就是500ms的延迟后会把数据返回给消费者

max.partition.fetch.bytes

服务器从每个分区里返回给消费者的最大字节数，默认值1MB；也就是说KafkaConsumer.poll()方法从每个分区里返回的记录不超过当前指定字节

举例: 如果一个主题有20个分区和5个消费者，那么每个消费者需要有4M的可用内存来接收记录，在为消费者分配内存时，可以给他们多分配一些，防止组内有消费者发生崩溃，这时其他的消费者就需要处理更多的分区。

注意事项：

1. max.partition.fetch.bytes的值必须必broker能够接收的最大消息的字节数(max.message.size)大，否则消费者可能无法读取这些消息，导致消费者一直挂起重试

2. 另一个需要考虑的因素是消费者处理数据的时间，消费者需要频繁调用poll()方法来避免会话过期和分区再均衡，如果返回的数据多，同时消费者需要很多时间来处理，就容易造成会话过期，那么要么将该值调小，要么延长会话过期时间

session.timeout.ms

消费者在被认为死亡之前可以与服务器断开连接的时间，默认值3s

如果在这个配置时间内消费者没有给群组协调器发送心跳，那就会被认为已经死亡，协调器就会触发再均衡

该属性和heartbeat.interval.ms（指定了poll()方法向协调器发送心跳的频率）紧密相关，而session.timeout.ms则指定了多久可以不发送心跳，这两个属性一般是同时配置，前者必须必后者小，一般是session.timeout.ms的三分之一

auto.offset.rest

该属性指定了消费者在读取一个没有偏移量的分区或者偏移量无效的情况下该如何处理，默认值是latest——也就是从最新的记录开始读取数据

另一个配置是earliest，意思是在偏移量无效的情况下，消费者将从起始位置读取分区记录

enable.auto.commit

是否开启自动提交，默认是true；但是可能出现重复数据和数据丢失，还可以通过配置auto.commit.interval.ms来控制提交频率

为了避免数据重复和丢失，可以将其设置为false，然后自己控制何时提交偏移量

partition.assignment.strategy

Range

该策略会把主题的若干个连续的分区分配给消费者，假设消费者C1和消费者C2同时订阅了主题T1和T2，并且每个主题有3个分区，那么消费者C1可能分配到这两个主题的分区0和分区1，C分配到分区2

RoundRobin

该策略会把主题的所有分区逐个分配给消费者，如果使用当前策略给消费者C1和C2来分配分区，那么C1将获得T1主题的分区0和分区2以及主题T2的分区1

默认使用org.apache.kafka.clients.consumer.RangeAssignor，这个类实现了Range策略；

也可以使用org.apache.kafka.clients.consumer.RoundRobinAssignor；

也可以自定义策略，配置值就是自定义的全类名

client.id

任意的字符串，用来标识从客户端发送过来的消息，通常被用在日志、度量指标和配额里

max.poll.records

该属性用于控制单词调用call()方法能够返回的记录数量，可以帮助控制在轮询里需要处理的数据量

receive.buffer.bytes & send.buffer.bytes

Socket在读写数据时用到的TCP缓冲也可以设置大小，如果被设置为-1就是使用操作系统默认值

问题和疑问汇总

容易混淆的

1.分区和副本的区别？

一个topic可以有多个分区，也可以有多个副本，本质上这俩属性没啥关系

假设有3个节点，5个分区，3个副本，那么分区可能分布在节点A(partition1,2) brokerB(partition,3,4,BrokerC(partition5))

那副本呢，就是每个partition都有3个备份，可能分布在不同的节点上

数据丢失的几种可能性

生产者写入时的数据丢失

1.提交策略不当，数据会丢失

2.提交时偏移量处理不正确，数据会丢失

3.关闭了不完全首领选举，acks=all，但是生产者写入数据时，首领挂了，正在重新选举中，Kafka会返回首领不可用的响应，但是生产者没能正确的处理——这不是broker可靠性问题，因为broker并没有收到这个消息，也不是一致性问题，因为消费者也没读到这个消息

Kafka接收数据时的数据丢失

1.副本还未同步，但是使用了首领接收即确认(acks=1)，

首领确认后宕机，数据会丢失

举例子：副本3个，禁用不完全首领选举，acks=1，生产者写入消息，首领接收并写入成功，返回“写入成功”响应，然后首领宕机，此时跟随副本还没有复制数据，但是他们会被认为是同步的（判断一个副本不同需要一小段时间），其中一个副本变成了新首领，因为消息并没有写入，所以数据丢失了

2.如果使用了不完全首领选举，数据可能会丢失

运维层的数据丢失

1.所有节点全部宕机，数据会丢失

2.acks=0

发送的对象无法序列化

网卡故障

分区离线或者整个集群长时间不可用

避免数据丢失的方法

注意事项

1.根据可靠性需求配置恰当的acks值

2.在参数配置和代码里正确的处理错误

核心概念

定义：基于发布与订阅的消息系统

主题和分区

消息：Kafka数据单元被称之为“消息”

Kafka的消息通过“主题”进行分类，一个主题可以跨多个服务器

“批次” ：就是属于同一个主题和分区的一组消息

主题可以被分为若干个分区，一个分区就是一个提交日志，Kafka通过分区来实现数据的冗余和伸缩，分区可以分布在不同的服务器

消息的写入与读取：消息以追加的方式写入分区然后以先入先出的顺序读取

消息的顺序：一个主题包含几个分区，因此无法在整个主题内保证消息的顺序，但是可以保证消息在单个分区内的顺序

流的概念：人们把一个主题的数据看成一个流，不管它有多少个分区，流是一组从生产者移动到消费者的数据

存储与传输

数据按一定的顺序持久化保存，可以按需读取

单个消息传输于网络，会导致较大的网络开销；消息分批传输可以减少网络开销，但是在时间和吞吐量上要做权衡，批次越大，单位时间内处理的消息就越多，单个消息的传输时间就越长，批次数据会被压缩，提升传输和存储能力，但要做更多的计算处理

Kafka的数据存储是基于二进制字节码的，晦涩难懂，所以有了Schema便于让人理解，通用的Avro作为序列化工具，同时也支持Json、String等，参考序列化类库和文档

Kafka是基于磁盘的数据存储，这也就是说Kafka允许消费者非实时的读取消息；消息被提交到磁盘，根据设置的保留规则进行保存，每个主题可以设置单独的保留规则，

生产者和消费者

Kafka的基本类型客户端：生产者和消费者；生产者创建消息，默认情况下，一个消息会被发布到一个特定的主题上，生产者在默认情况下把消息均衡的分布到主题的所有分区上，而不关心特定消息会被写到哪个分区，不过也可以用分区器将其映射到指定的分区；

更高级的客户端：用于数据集成的Kafka Connect API和用于流式处理的Kafka Streams；

多个生产者： kafka可以无缝的支持多个生产者，不管客户端在使用单个主题还是多个主题

多个消费者： Kafka支持多个消费者从一个单独的消息流上读取数据，而且消费者之间互不影响；其他的消息队列的消息一旦被一个客户端读取，其他客户端就无法再读取它。多个消费者可以组成一个群组，他们共享一个消息流，并保证整个群组对每个给定的消息只处理一次。

消费者群组：消费者是消费者群组的一部分，也就是说会有一个或多个消费者共同读取一个主题；一个群组里的消费者订阅的是同一个主题，每个消费者接收主题一部分分区的消息（消费者的个数一定小于等于分区数）

消费所有权：群组保证每个分区只能被一个消费者使用，消费者与分区之间的映射通常被称为消费者对分区的所有权关系

偏移量：他是一个不断递增的整数值，在创建消息时，kafka会把它添加到消息里，在给定的分区里，每个消息的偏移量都是唯一的，消费者把每个分区最后读取的消息偏移量保存在Zookeeper或kafka上，如果消费者关闭或者重启，它的读取状态不会丢失

Broker和集群

一个独立的kafka服务器被称为broker

broker接收来自生产者的消息，为消息设置偏移量，并提交消息到磁盘保存；

Broker为消费者提供服务，对读取分区的请求做出响应，返回已经提交到磁盘上的消息

broker是集群的组成部分，每个集群都有一个broker同时充当了“集群控制器”的角色（自动从活跃成员中选举出来）

在集群中一个分区从属于一个broker，该broker被称为分区的首领；一个分区可以分配给多个broker，这时候会发生分区复制

保留消息：是Kafka的一个重要特性，Kafka默认的消息保留策略是：要么保留一段时间(比如7天)，要么保留到消息达到一定大小的字节数（比如10GB），如果达到任意一个阀值，旧消息就会被删除。

多集群：随着Kafka数量的增加，基于以下几点原因，最好使用多个集群：1.数据类型分离 2.安全需求隔离 3.多数据中心（灾难恢复）

多集群直接的消息复制： Kafka提供了一个叫做MirrorMaker的工具，可以用它来实现集群间的消息复制，其核心组件包含了一个生产者和一个消费者，两者之间通过一个队列相连。

为什么选择Kafka

1.多个生产者： Kafka可以无缝的支持多个生产者，不管客户端在使用单个主题还是多个主题。所以它适合用来从多个前端系统收集数据，并以统一的格式对外提供数据

2.多个消费者： kafka支持多个消费者从一个单独的消息流上读取数据，而且消费者之间互不影响；多个消费者可以组成一个群组，它们共享一个消息流，并保证整个群组对每个给定的消息只处理一次。其它队列系统的消息一旦被一个客户端读取，其它客户端就无法再读取它。

3.基于磁盘的数据存储： Kafka允许消费者非实时的读取消息，这归功于Kafka的数据保留特性，消息被提交到磁盘，根据设置的保留规则进行保存；每个主题可以设置单独的保留规则，以便满足不同消费者的需求，各个主题可以保留不同数量的消息。

4.伸缩性： Kafka可以单点提供服务，也可以集群提供服务，甚至上百个broker

5.高性能：通过横向扩展生产者、消费者、broker，kafka可以轻松处理巨大的消息流，在处理数据的同时，他还能保证亚秒级的消息延迟

Kafka Producer

创建生产者

发送消息到kafka

同步发送消息

异步发送消息

生产者的配置

分区

序列化

Kafka的可靠性

可靠性保证

1.Kafka可以保证分区消息的顺序；如果同一个生产者往同一个分区写入消息，而且消息B在消息A之后写入，那么Kafka是可以保证消息B的偏移量一定比消息A的偏移量大

2.消息的“已提交”

只有当消息被写入分区的所有同步副本时（但不一定要写入磁盘），它才被认为是“已提交”的

消息的确认；不同类型确认

消息被完全提交时确认

acks=all

与min.insync.replicas参数结合，决定返回确认时要至少有多个副本能够接到消息，这是最保险的，也是最慢的

上述方法想加速可以使用异步和大批次来加速，但是还是会降低吞吐量

消息被写入首领副本时的确认

acks=1

首领在收到消息并把它写入到分区数据文件时，返回确认或者错误

这时数据是否丢失，取决于生产者能否正确处理恰好碰到首领选举时LeaderNotAvailableException异常

以及如果遇到了首领确认后，未复制给跟随节点就宕机了，也会丢失数据

消息被发送到网络时的确认

acks=0；如果生产者通过网络把消息发送出去，就认为消息已经写入kafka

数据丢失

发送的对象无法序列化

网卡故障

分区离线或者整个集群长时间不可用

基准测试

运行速度很快，所以经常使用这种模式做基准测试

3.只要有一个副本是活跃的，那么已经提交的消息就不会丢失

4.消费者只能读取已提交的消息

复制

Kafka的复制机制和分区的多副本架构是Kafka可靠性保证的核心

副本的概念

首领副本

1.接受事件请求

跟随副本

1.保持和首领副本同步，但是不一定每个副本都会同步，比如出现异常

2.首领副本故障会从跟随副本中选举新的

同步副本

数据最实时更新的，已经更新的副本，并且是存活状态

首领副本或者的时候，它一定是同步的，跟随副本正常情况下也是同步副本，但是出现网络延迟、挂掉，就都不是同步副本

复习:主题—>多个分区—>分区内有序—>一个分区多个副本—>其中一个副本是首领—>所有的事件都直接发送给首领或者直接从首领副本读取事件—>其他副本只需要和首领副本保持同步—>首领副本故障，重新选举一个新的副本作为首领

跟随者副本需要满足的条件

1.与Zookeeper之间有一个活跃的会话，也就是说，它在过去的6s内向Zookeeper发送过心跳

2.在过去的10s内从首领那里获取过信息

3.在过去的10s内从首领获取过最新的消息；光从首领获取消息是不够的，它还必须是几乎零延迟的

end：如果不能满足上述的任何一点，就认为它是不同步的，这是就需要重新与zookeeper重新建立连接

supplement：如果一个或多个副本在同步和非同步状态之间快速切换，说明集群内部出现了问题，通常是java不恰当的GC导致；不恰当的GC配置会导致几秒的停顿，从而让Broker和zookeeper断开连接

性能影响：一个滞后的同步副本会导致生产者和消费者变慢；如果副本不需要同步了那么对性能无影响

broker配置

可靠性配置:

Kafka有3个配置参数会影响Kafka消息存储的可靠性

配置可以应用在broker级别，用于控制所有主题的行为

配置可以应用在主题级别，用于控制个别主题行为

配置细节

1.复制系数

主题级别的配置参数是replication.factor

broker级别的，可以通过default.replication.factor来配置自动创建的主题

复制系数小于等于broker节点数，同时具有相对应的副本数，也就是会占用对应N倍的磁盘空间

复制系数如果为1会出现broker重启或者停止的时候，主题不可用

如果复制系数为2，表面看是可以容忍1个broker失效，但是这时候还可能导致集群不稳定，迫使重启另一个broker——集群控制器，这时候其实服务也是不可用的

默认的复制系数为3

容灾的再思考，如何分布broker

不同的主机？不同的机架（broker.rack）?不同的地域

2.不完全的首领选举

问题产生的原因

场景一

分区有3个副本，首先跟随副本都不可用，这时生产者继续向首领写入数据，然后首领也不可用了，这时，如果之前的跟随者重启，它就成了分区的唯一不同步副本

场景二

分区有3个副本，因为网络问题，两个跟随副本复制消息滞后，所以他们尽管在复制，但是已经不同步了，首领做为同步副本继续接收，这是如果首领不可用，另外两个副本也就无法变成同步了

两难的选择

1.数据丢失和数据不一致？

如果不同的副本可以被提升为新首领，那么在这个副本变为不同步后写入旧首领的消息会全部丢失，导致数据不一致

2.服务一段时间的不可用？

如果不同步的副本不能被提升为新首领，那么分区在旧首领恢复之前是不可用的

配置: unclean.leader.election.enable=true/false

true 允许新副本成为首领，需要承担数据丢失风险

false 对数据的一致性有高质量的要求，需要承担服务不可用的风险——可用性降低

3.最少同步副本

在主题级别和broker级别，这个参数都叫min.insync.replicas

举个例子：如果当前参数配置为2，然后一共有3个broker；那么如果一个跟随副本挂了，另一个跟随副本还能正常同步，此时系统还是OK的，因为满足配置，但是如果2个跟随副本都挂了，此时就不满足配置，生产者再继续写入数据就会收到NotEnoughReplicasException,但是消费者是可以读数据的，也就是说此时整个集群变成了只读的，什么时候恢复呢，也就是让另外两个或者有新的节点恢复成可用的。

在可靠的系统里使用生产者

不同的确认模式的异同

数据丢失

生产者处理不当

确认模式带来的数据丢失

生产者的重试次数

生产者需要处理的错误

1.可以自动处理的错误

2.需要手动处理的错误

在可靠的系统里使用消费者

验证系统可靠性

配置验证