初识Kafka

Kafka之所以受到越来越多的青睐，与它所“扮演”的三大角色是分不开的：

• 消息系统：Kafka 和传统的消息系统（也称作消息中间件）都具备系统解耦、冗余存储、流量削峰、缓冲、异步通信、扩展性、可恢复性等功能。与此同时，Kafka 还提供了大多数消息系统难以实现的消息顺序性保障及回溯消费的功能。
• 存储系统：Kafka 把消息持久化到磁盘，相比于其他基于内存存储的系统而言，有效地降低了数据丢失的风险。也正是得益于Kafka 的消息持久化功能和多副本机制，我们可以把Kafka作为长期的数据存储系统来使用，只需要把对应的数据保留策略设置为“永久”或启用主题的日志压缩功能即可。
• 流式处理平台：Kafka 不仅为每个流行的流式处理框架提供了可靠的数据来源，还提供了一个完整的流式处理类库，比如窗口、连接、变换和聚合等各类操作。

基本概念

一个典型的 Kafka 体系架构包括若干 Producer、若干 Broker、若干 Consumer，以及一个ZooKeeper集群。其中ZooKeeper是Kafka用来负责集群元数据的管理、控制器的选举等操作的。Producer将消息发送到Broker，Broker负责将收到的消息存储到磁盘中，而Consumer负责从Broker订阅并消费消息。整个Kafka体系结构中引入了以下3个术语：

• Producer：生产者，也就是发送消息的一方。生产者负责创建消息，然后将其投递到Kafka中。
• Consumer：消费者，也就是接收消息的一方。消费者连接到Kafka上并接收消息，进而进行相应的业务逻辑处理。
• Broker：服务代理节点。对于Kafka而言，Broker可以简单地看作一个独立的Kafka服务节点或Kafka服务实例。大多数情况下也可以将Broker看作一台Kafka服务器，前提是这台服务器上只部署了一个Kafka实例。一个或多个Broker组成了一个Kafka集群。一般而言，我们更习惯使用首字母小写的broker来表示服务代理节点。

在Kafka中还有两个特别重要的概念—**主题（Topic）与分区（Partition）**​。Kafka中的消息以主题为单位进行归类，生产者负责将消息发送到特定的主题（发送到Kafka集群中的每一条消息都要指定一个主题）​，而消费者负责订阅主题并进行消费。

主题是一个逻辑上的概念，它还可以细分为多个分区，一个分区只属于单个主题，很多时候也会把分区称为主题分区（Topic-Partition）​。同一主题下的不同分区包含的消息是不同的，分区在存储层面可以看作一个可追加的日志（Log）文件，消息在被追加到分区日志文件的时候都会分配一个特定的偏移量（offset）​。offset是消息在分区中的唯一标识，Kafka通过它来保证消息在分区内的顺序性，不过offset并不跨越分区，也就是说，Kafka保证的是分区有序而不是主题有序。

Kafka中的分区可以分布在不同的服务器（broker）上，也就是说，一个主题可以横跨多个broker，以此来提供比单个broker更强大的性能。

每一条消息被发送到broker之前，会根据分区规则选择存储到哪个具体的分区。如果分区规则设定得合理，所有的消息都可以均匀地分配到不同的分区中。

Kafka 为分区引入了多副本（Replica）机制，通过增加副本数量可以提升容灾能力。同一分区的不同副本中保存的是相同的消息（在同一时刻，副本之间并非完全一样）​，副本之间是“一主多从”的关系，其中leader副本负责处理读写请求，follower副本只负责与leader副本的消息同步。副本处于不同的broker中，当leader副本出现故障时，从follower副本中重新选举新的leader副本对外提供服务。Kafka通过多副本机制实现了故障的自动转移，当Kafka集群中某个broker失效时仍然能保证服务可用。

生产者和消费者只与leader副本进行交互，而follower副本只负责消息的同步，很多时候follower副本中的消息相对leader副本而言会有一定的滞后。

Kafka 消费端也具备一定的容灾能力。Consumer 使用拉（Pull）模式从服务端拉取消息，并且保存消费的具体位置，当消费者宕机后恢复上线时可以根据之前保存的消费位置重新拉取需要的消息进行消费，这样就不会造成消息丢失。

分区中的所有副本统称为AR（Assigned Replicas）​。所有与leader副本保持一定程度同步的副本（包括leader副本在内）组成ISR（In-Sync Replicas）​，ISR集合是AR集合中的一个子集。消息会先发送到leader副本，然后follower副本才能从leader副本中拉取消息进行同步，同步期间内follower副本相对于leader副本而言会有一定程度的滞后。前面所说的“一定程度的同步”是指可忍受的滞后范围，这个范围可以通过参数进行配置。与leader副本同步滞后过多的副本（不包括leader副本）组成**OSR（Out-of-Sync Replicas）**​，由此可见，AR=ISR+OSR。在正常情况下，所有的follower 副本都应该与 leader 副本保持一定程度的同步，即AR=ISR，OSR集合为空。

leader副本负责维护和跟踪ISR集合中所有follower副本的滞后状态，当follower副本落后太多或失效时，leader副本会把它从ISR集合中剔除。如果OSR集合中有follower副本“追上”了leader副本，那么leader副本会把它从OSR集合转移至ISR集合。默认情况下，当leader副本发生故障时，只有在ISR集合中的副本才有资格被选举为新的leader，而在OSR集合中的副本则没有任何机会（不过这个原则也可以通过修改相应的参数配置来改变）​。

ISR与HW和LEO也有紧密的关系。HW是High Watermark的缩写，俗称高水位，它标识了一个特定的消息偏移量（offset）​，消费者只能拉取到这个offset之前的消息。

LEO是Log End Offset的缩写，它标识当前日志文件中下一条待写入消息的offset，LEO的大小相当于当前日志分区中最后一条消息的offset值加1。分区ISR集合中的每个副本都会维护自身的LEO，而ISR集合中最小的LEO即为分区的HW，对消费者而言只能消费HW之前的消息。（说白了就是没同步的消息还不允许被消费）

由此可见，Kafka 的复制机制既不是完全的同步复制，也不是单纯的异步复制。事实上，同步复制要求所有能工作的 follower 副本都复制完，这条消息才会被确认为已成功提交，这种复制方式极大地影响了性能。而在异步复制方式下，follower副本异步地从leader副本中复制数据，数据只要被leader副本写入就被认为已经成功提交。在这种情况下，如果follower副本都还没有复制完而落后于leader副本，突然leader副本宕机，则会造成数据丢失。Kafka使用的这种ISR的方式则有效地权衡了数据可靠性和性能之间的关系。

安装与配置

ZooKeeper安装与配置

ZooKeeper是安装Kafka集群的必要组件，Kafka通过ZooKeeper来实施对元数据信息的管理，包括集群、broker、主题、分区等内容。（不过新版本kafka已经摆脱了zk的依赖，而是把元数据存储到了特定的topic里面）

Kafka的安装与配置

修改broker的配置文件$KAFKA_HOME/conf/server.properties。主要关注以下几个配置参数即可：
如果是单机模式，那么修改完上述配置参数之后就可以启动服务。如果是集群模式，那么只需要对单机模式的配置文件做相应的修改即可：确保集群中每个broker的broker.id配置参数的值不一样，以及listeners配置参数也需要修改为与broker对应的IP地址或域名，之后就可以各自启动服务。注意，在启动 Kafka 服务之前同样需要确保 zookeeper.connect参数所配置的ZooKeeper服务已经正确启动。

生产与消费

Kafka提供了许多实用的脚本工具，存放在$KAFKA_HOME的bin目录下，其中与主题有关的就是 kafka-topics.sh 脚本，下面我们用它演示创建一个分区数为 4、副本因子为 3 的主题topic-demo，示例如下：

其中–zookeeper指定了Kafka所连接的ZooKeeper服务地址，–topic指定了所要创建主题的名称，–replication-factor 指定了副本因子，–partitions 指定了分区个数，–create是创建主题的动作指令。

服务端参数配置

下面挑选一些重要的服务端参数来做细致的说明，这些参数都配置在$KAFKA_HOME/config/server.properties文件中。

1. zookeeper.connect：该参数指明broker要连接的ZooKeeper集群的服务地址（包含端口号）​，没有默认值，且此参数为必填项。可以配置为localhost：2181，如果ZooKeeper集群中有多个节点，则可以用逗号将每个节点隔开，类似于 localhost1：2181，localhost2：2181，localhost3：2181这种格式。最佳的实践方式是再加一个chroot路径，这样既可以明确指明该chroot路径下的节点是为Kafka所用的，也可以实现多个Kafka集群复用一套ZooKeeper集群，这样可以节省更多的硬件资源。包含 chroot 路径的配置类似于 localhost1：2181，localhost2：2181，localhost3：2181/kafka这种，如果不指定chroot，那么默认使用ZooKeeper的根路径。
2. listeners：该参数指明broker监听客户端连接的地址列表，即为客户端要连接broker的入口地址列表，配置格式为protocol1：//hostname1：port1，protocol2：//hostname2：port2，其中protocol代表协议类型，Kafka当前支持的协议类型有PLAINTEXT、SSL、SASL_SSL等，如果未开启安全认证，则使用简单的PLAINTEXT即可。hostname代表主机名，port代表服务端口，此参数的默认值为null
3. broker.id：该参数用来指定Kafka集群中broker的唯一标识，默认值为-1。如果没有设置，那么Kafka会自动生成一个。这个参数还和meta.properties文件及服务端参数broker.id.generation.enable和reserved.broker.max.id有关
4. log.dir和log.dirs：Kafka 把所有的消息都保存在磁盘上，而这两个参数用来配置Kafka 日志文件存放的根目录。一般情况下，log.dir 用来配置单个根目录，而 log.dirs 用来配置多个根目录（以逗号分隔）​，但是Kafka并没有对此做强制性限制，也就是说，log.dir和log.dirs都可以用来配置单个或多个根目录。log.dirs 的优先级比 log.dir 高，但是如果没有配置log.dirs，则会以 log.dir 配置为准。默认情况下只配置了 log.dir 参数，其默认值为/tmp/kafka-logs。
5. message.max.bytes：该参数用来指定broker所能接收消息的最大值，默认值为1000012（B）​，约等于976.6KB。如果 Producer 发送的消息大于这个参数所设置的值，那么（Producer）就会报出RecordTooLargeException的异常。如果需要修改这个参数，那么还要考虑max.request.size （客户端参数）​、max.message.bytes（topic端参数）等参数的影响。为了避免修改此参数而引起级联的影响，建议在修改此参数之前考虑分拆消息的可行性。

生产者

客户端开发

消息对象 ProducerRecord

这里有必要单独说明的是构建的消息对象 ProducerRecord，它并不是单纯意义上的消息，它包含了多个属性，原本需要发送的与业务相关的消息体只是其中的一个 value 属性，比如“Hello，Kafka！”只是ProducerRecord对象中的一个属性。ProducerRecord类的定义如下（只截取成员变量）​：

其中topic和partition字段分别代表消息要发往的主题和分区号。headers字段是消息的头部，Kafka 0.11.x版本才引入这个属性，它大多用来设定一些与应用相关的信息，如无需要也可以不用设置。key是用来指定消息的键，它不仅是消息的附加信息，还可以用来计算分区号进而可以让消息发往特定的分区。前面提及消息以主题为单位进行归类，而这个key可以让消息再进行二次归类，同一个key的消息会被划分到同一个分区中，有key的消息还可以支持日志压缩的功能。value是指消息体，一般不为空，如果为空则表示特定的消息—墓碑消息。timestamp是指消息的时间戳，它有CreateTime和LogAppendTime两种类型，前者表示消息创建的时间，后者表示消息追加到日志文件的时间。

必要的参数配置

在Kafka生产者客户端KafkaProducer中有3个参数是必填的：

1. bootstrap.servers：该参数用来指定生产者客户端连接Kafka集群所需的broker地址清单，具体的内容格式为host1：port1，host2：port2，可以设置一个或多个地址，中间以逗号隔开，此参数的默认值为“​”​。注意这里并非需要所有的broker地址，因为生产者会从给定的broker里查找到其他broker的信息。不过建议至少要设置两个以上的broker 地址信息，当其中任意一个宕机时，生产者仍然可以连接到 Kafka集群上。
2. key.serializer 和 value.serializer：broker 端接收的消息必须以字节数组（byte[​]​）的形式存在。key.serializer和value.serializer这两个参数分别用来指定key和value序列化操作的序列化器，这两个参数无默认值。注意这里必须填写序列化器的全限定名，如org.apache.kafka.common.serialization.StringSerializer。

KafkaProducer中的参数众多，我们可以直接使用客户端中的org.apache.kafka.clients.producer.ProducerConfig类来做一定程度上的预防措施，每个参数在 ProducerConfig 类中都有对应的名称。
KafkaProducer是线程安全的，可以在多个线程中共享单个KafkaProducer实例，也可以将KafkaProducer实例进行池化来供其他线程调用。

发送消息

创建生产者实例和构建消息之后，就可以开始发送消息了。发送消息主要有三种模式：**发后即忘（fire-and-forget）​、同步（sync）及异步（async）**​。
KafkaProducer 的 send（​）方法并非是 void 类型，而是 Future＜RecordMetadata＞类型，send（​）方法有2个重载方法，具体定义如下：

KafkaProducer中一般会发生两种类型的异常：可重试的异常和不可重试的异常。对于可重试的异常，如果配置了 retries 参数，那么只要在规定的重试次数内自行恢复了，就不会抛出异常。retries参数的默认值为0。

同步发送的方式可靠性高，要么消息被发送成功，要么发生异常。如果发生异常，则可以捕获并进行相应的处理，而不会像“发后即忘”的方式直接造成消息的丢失。不过同步发送的方式的性能会差很多，需要阻塞等待一条消息发送完之后才能发送下一条。

我们再来了解一下异步发送的方式，一般是在send（​）方法里指定一个Callback的回调函数，Kafka在返回响应时调用该函数来实现异步的发送确认。有读者或许会有疑问，send（​）方法的返回值类型就是Future，而Future本身就可以用作异步的逻辑处理。这样做不是不行，只不过Future里的 get（​）方法在何时调用，以及怎么调用都是需要面对的问题，消息不停地发送，那么诸多消息对应的Future对象的处理难免会引起代码处理逻辑的混乱。使用Callback的方式非常简洁明了，Kafka有响应时就会回调，要么发送成功，要么抛出异常。

对于同一个分区而言，如果消息record1于record2之前先发送（参考上面的示例代码）​，那么KafkaProducer就可以保证对应的callback1在callback2之前调用，也就是说，回调函数的调用也可以保证分区有序。

通常，一个KafkaProducer不会只负责发送单条消息，更多的是发送多条消息，在发送完这些消息之后，需要调用KafkaProducer的close（​）方法来回收资源。close（​）方法会阻塞等待之前所有的发送请求完成后再关闭KafkaProducer。与此同时，KafkaProducer还提供了一个带超时时间的close（​）方法。

序列化

生产者需要用序列化器（Serializer）把对象转换成字节数组才能通过网络发送给Kafka。而在对侧，消费者需要用反序列化器（Deserializer）把从 Kafka 中收到的字节数组转换成相应的对象（所以说如果生产者和消费者序列化方式不一样，就无法正确的解析消息）。

如果 Kafka 客户端提供的几种序列化器都无法满足应用需求，则可以选择使用如 Avro、JSON、Thrift、ProtoBuf和Protostuff等通用的序列化工具来实现，或者使用自定义类型的序列化器来实现。

分区器

消息在通过send（​）方法发往broker的过程中，有可能需要经过拦截器（Interceptor）​、序列化器（Serializer）和分区器（Partitioner）的一系列作用之后才能被真正地发往 broker。消息经过序列化之后就需要确定它发往的分区，如果消息ProducerRecord中指定了partition字段，那么就不需要分区器的作用，因为partition代表的就是所要发往的分区号。

如果消息ProducerRecord中没有指定partition字段，那么就需要依赖分区器，根据key这个字段来计算partition的值，分区器的作用就是为消息分配分区。

Kafka中提供的默认分区器是org.apache.kafka.clients.producer.internals.DefaultPartitioner，它实现了org.apache.kafka.clients.producer.Partitioner接口，如图所示：

其中partition（​）方法用来计算分区号，返回值为int类型。partition（​）方法中的参数分别表示主题、键、序列化后的键、值、序列化后的值，以及集群的元数据信息，通过这些信息可以实现功能丰富的分区器。close（​）方法在关闭分区器的时候用来回收一些资源。

在默认分区器 DefaultPartitioner 的实现中，close（​）是空方法，而在 partition（​）方法中定义了主要的分区分配逻辑。如果key 不为 null，那么默认的分区器会对 key 进行哈希（采用MurmurHash2算法，具备高运算性能及低碰撞率）​，最终根据得到的哈希值来计算分区号，拥有相同key的消息会被写入同一个分区。如果key为null，那么消息将会以轮询的方式发往主题内的各个可用分区。

在不改变主题分区数量的情况下，key与分区之间的映射可以保持不变。不过，一旦主题中增加了分区，那么就难以保证key与分区之间的映射关系了。

了使用 Kafka 提供的默认分区器进行分区分配，还可以使用自定义的分区器，只需同DefaultPartitioner一样实现Partitioner接口即可。默认的分区器在key为null时不会选择非可用的分区，我们可以通过自定义的分区器DemoPartitioner来打破这一限制。

生产者拦截器

生产者拦截器既可以用来在消息发送前做一些准备工作，比如按照某个规则过滤不符合要求的消息、修改消息的内容等，也可以用来在发送回调逻辑前做一些定制化的需求，比如统计类工作。

生产者拦截器的使用也很方便，主要是自定义实现org.apache.kafka.clients.producer.ProducerInterceptor接口。ProducerInterceptor接口中包含3个方法。KafkaProducer在将消息序列化和计算分区之前会调用生产者拦截器的onSend（​）方法来对消息进行相应的定制化操作，KafkaProducer 会在消息被应答（Acknowledgement）之前或消息发送失败时调用生产者拦截器的 onAcknowledgement（​）方法，优先于用户设定的 Callback 之前执行。

KafkaProducer中不仅可以指定一个拦截器，还可以指定多个拦截器以形成拦截链。拦截链会按照 interceptor.classes 参数配置的拦截器的顺序来一一执行（配置的时候，各个拦截器之间使用逗号隔开）​。

在拦截链中，如果某个拦截器执行失败，那么下一个拦截器会接着从上一个执行成功的拦截器继续执行。（拦截器失败一个不影响其他的执行）

原理分析

整体架构

生产者客户端的整体架构图：

整个生产者客户端由两个线程协调运行，这两个线程分别为主线程和Sender线程（发送线程）​。在主线程中由KafkaProducer创建消息，然后通过可能的拦截器、序列化器和分区器的作用之后缓存到消息累加器（RecordAccumulator，也称为消息收集器） 中。Sender 线程负责从RecordAccumulator中获取消息并将其发送到Kafka中。

RecordAccumulator 主要用来缓存消息以便 Sender 线程可以批量发送，进而减少网络传输的资源消耗以提升性能。RecordAccumulator 缓存的大小可以通过生产者客户端参数buffer.memory 配置，默认值为 33554432B，即 32MB。如果生产者发送消息的速度超过发送到服务器的速度，则会导致生产者空间不足，这个时候KafkaProducer的send（​）方法调用要么被阻塞，要么抛出异常，这个取决于参数max.block.ms的配置，此参数的默认值为60000，即60秒。

主线程中发送过来的消息都会被追加到RecordAccumulator的某个双端队列（Deque）中，在 RecordAccumulator 的内部为每个分区都维护了一个双端队列，队列中的内容就是ProducerBatch，即 Deque＜ProducerBatch＞。消息写入缓存时，追加到双端队列的尾部；Sender读取消息时，从双端队列的头部读取。注意ProducerBatch不是ProducerRecord，ProducerBatch中可以包含一至多个 ProducerRecord。通俗地说，ProducerRecord 是生产者中创建的消息，而ProducerBatch是指一个消息批次，ProducerRecord会被包含在ProducerBatch中，这样可以使字节的使用更加紧凑。与此同时，将较小的ProducerRecord拼凑成一个较大的ProducerBatch，也可以减少网络请求的次数以提升整体的吞吐量。ProducerBatch和消息的具体格式有关，如果生产者客户端需要向很多分区发送消息，则可以将buffer.memory参数适当调大以增加整体的吞吐量。

消息在网络上都是以字节（Byte）的形式传输的，在发送之前需要创建一块内存区域来保存对应的消息。在Kafka生产者客户端中，通过java.io.ByteBuffer实现消息内存的创建和释放。不过频繁的创建和释放是比较耗费资源的，在RecordAccumulator的内部还有一个BufferPool，它主要用来实现ByteBuffer的复用，以实现缓存的高效利用。不过BufferPool只针对特定大小的ByteBuffer进行管理，而其他大小的ByteBuffer不会缓存进BufferPool中，这个特定的大小由batch.size参数来指定，默认值为16384B，即16KB。我们可以适当地调大batch.size参数以便多缓存一些消息。

Sender 从 RecordAccumulator 中获取缓存的消息之后，会进一步将原本＜分区，Deque＜ProducerBatch＞＞的保存形式转变成＜Node，List＜ ProducerBatch＞的形式，其中Node表示Kafka集群的broker节点。对于网络连接来说，生产者客户端是与具体的broker节点建立的连接，也就是向具体的 broker 节点发送消息，而并不关心消息属于哪一个分区；而对于 KafkaProducer的应用逻辑而言，我们只关注向哪个分区中发送哪些消息，所以在这里需要做一个应用逻辑层面到网络I/O层面的转换。

在转换成＜Node，List＜ProducerBatch＞＞的形式之后，Sender 还会进一步封装成＜Node，Request＞的形式，这样就可以将Request请求发往各个Node了，这里的Request是指Kafka的各种协议请求，对于消息发送而言就是指具体的ProduceRequest。

请求在从Sender线程发往Kafka之前还会保存到InFlightRequests中，InFlightRequests保存对象的具体形式为 Map＜NodeId，Deque＜Request＞＞，它的主要作用是缓存了已经发出去但还没有收到响应的请求（NodeId 是一个 String 类型，表示节点的 id 编号）​。与此同时，InFlightRequests还提供了许多管理类的方法，并且通过配置参数还可以限制每个连接（也就是客户端与Node之间的连接）最多缓存的请求数。这个配置参数为max.in.flight.requests.per.connection，默认值为 5，即每个连接最多只能缓存 5 个未响应的请求，超过该数值之后就不能再向这个连接发送更多的请求了，除非有缓存的请求收到了响应（Response）​。通过比较Deque＜Request＞的size与这个参数的大小来判断对应的Node中是否已经堆积了很多未响应的消息，如果真是如此，那么说明这个 Node 节点负载较大或网络连接有问题，再继续向其发送请求会增大请求超时的可能。

元数据的更新

InFlightRequests还可以获得leastLoadedNode，即所有Node中负载最小的那一个。这里的负载最小是通过每个Node在InFlightRequests中还未确认的请求决定的，未确认的请求越多则认为负载越大。对于图中的InFlightRequests 来说，图中展示了三个节点Node0、Node1和Node2，很明显Node1的负载最小。也就是说，Node1为当前的leastLoadedNode。选择leastLoadedNode发送请求可以使它能够尽快发出，避免因网络拥塞等异常而影响整体的进度。leastLoadedNode的概念可以用于多个应用场合，比如元数据请求、消费者组播协议的交互。

KafkaProducer要将此消息追加到指定主题的某个分区所对应的leader副本之前，首先需要知道主题的分区数量，然后经过计算得出（或者直接指定）目标分区，之后KafkaProducer需要知道目标分区的leader副本所在的broker 节点的地址、端口等信息才能建立连接，最终才能将消息发送到 Kafka，在这一过程中所需要的信息都属于元数据信息。

我们了解了bootstrap.servers参数只需要配置部分broker节点的地址即可，不需要配置所有broker节点的地址，因为客户端可以自己发现其他broker节点的地址，这一过程也属于元数据相关的更新操作。与此同时，分区数量及leader副本的分布都会动态地变化，客户端也需要动态地捕捉这些变化。

元数据是指Kafka集群的元数据，这些元数据具体记录了集群中有哪些主题，这些主题有哪些分区，每个分区的leader副本分配在哪个节点上，follower副本分配在哪些节点上，哪些副本在AR、ISR等集合中，集群中有哪些节点，控制器节点又是哪一个等信息。

当客户端中没有需要使用的元数据信息时，比如没有指定的主题信息，或者超过metadata.max.age.ms 时间没有更新元数据都会引起元数据的更新操作。客户端参数metadata.max.age.ms的默认值为300000，即5分钟。元数据的更新操作是在客户端内部进行的，对客户端的外部使用者不可见。当需要更新元数据时，会先挑选出leastLoadedNode，然后向这个Node发送MetadataRequest请求来获取具体的元数据信息。这个更新操作是由Sender线程发起的，在创建完MetadataRequest之后同样会存入InFlightRequests，之后的步骤就和发送消息时的类似。元数据虽然由Sender线程负责更新，但是主线程也需要读取这些信息，这里的数据同步通过synchronized和final关键字来保障。

重要的生产者参数

下面挑选一些重要的参数进行讲解:

acks

这个参数用来指定分区中必须要有多少个副本收到这条消息，之后生产者才会认为这条消息是成功写入的。acks 是生产者客户端中一个非常重要的参数，它涉及消息的可靠性和吞吐量之间的权衡。acks参数有3种类型的值（都是字符串类型）​。

• acks=1。默认值即为1。生产者发送消息之后，只要分区的leader副本成功写入消息，那么它就会收到来自服务端的成功响应。如果消息无法写入leader副本，比如在leader 副本崩溃、重新选举新的 leader 副本的过程中，那么生产者就会收到一个错误的响应，为了避免消息丢失，生产者可以选择重发消息。如果消息写入leader副本并返回成功响应给生产者，且在被其他follower副本拉取之前leader副本崩溃，那么此时消息还是会丢失，因为新选举的leader副本中并没有这条对应的消息。acks设置为1，是消息可靠性和吞吐量之间的折中方案。
• acks=0。生产者发送消息之后不需要等待任何服务端的响应。如果在消息从发送到写入Kafka的过程中出现某些异常，导致Kafka并没有收到这条消息，那么生产者也无从得知，消息也就丢失了。在其他配置环境相同的情况下，acks 设置为 0 可以达到最大的吞吐量。
• acks=-1或acks=all。生产者在消息发送之后，需要等待ISR中的所有副本都成功写入消息之后才能够收到来自服务端的成功响应。在其他配置环境相同的情况下，acks 设置为-1（all）可以达到最强的可靠性。但这并不意味着消息就一定可靠，因为ISR中可能只有leader副本，这样就退化成了acks=1的情况。要获得更高的消息可靠性需要配合 min.insync.replicas 等参数的联动。

max.request.size

这个参数用来限制生产者客户端能发送的消息的最大值，默认值为1048576B，即 1MB。一般情况下，这个默认值就可以满足大多数的应用场景了。并不建议盲目地增大这个参数的配置值，尤其是在对Kafka整体脉络没有足够把控的时候。因为这个参数还涉及一些其他参数的联动，比如broker端的message.max.bytes参数，如果配置错误可能会引起一些不必要的异常。比如将broker端的message.max.bytes参数配置为10，而max.request.size参数配置为20，那么当我们发送一条大小为15B的消息时，生产者客户端就会报出异常。

retries和retry.backoff.ms

retries参数用来配置生产者重试的次数，默认值为0，即在发生异常的时候不进行任何重试动作。消息在从生产者发出到成功写入服务器之前可能发生一些临时性的异常，比如网络抖动、leader副本的选举等，这种异常往往是可以自行恢复的，生产者可以通过配置retries大于0的值，以此通过内部重试来恢复而不是一味地将异常抛给生产者的应用程序。如果重试达到设定的次数，那么生产者就会放弃重试并返回异常。不过并不是所有的异常都是可以通过重试来解决的，比如消息太大，超过max.request.size参数配置的值时，这种方式就不可行了。

重试还和另一个参数retry.backoff.ms有关，这个参数的默认值为100，它用来设定两次重试之间的时间间隔，避免无效的频繁重试。在配置 retries 和 retry.backoff.ms之前，最好先估算一下可能的异常恢复时间，这样可以设定总的重试时间大于这个异常恢复时间，以此来避免生产者过早地放弃重试。

如果将acks参数配置为非零值，并且max.in.flight.requests.per.connection参数配置为大于1的值，那么就会出现错序的现象：如果第一批次消息写入失败，而第二批次消息写入成功，那么生产者会重试发送第一批次的消息，此时如果第一批次的消息写入成功，那么这两个批次的消息就出现了错序。一般而言，在需要保证消息顺序的场合建议把参数max.in.flight.requests.per.connection配置为1，而不是把acks配置为0，不过这样也会影响整体的吞吐。

compression.type

这个参数用来指定消息的压缩方式，默认值为“none”​，即默认情况下，消息不会被压缩。该参数还可以配置为“gzip”​“snappy”和“lz4”​。对消息进行压缩可以极大地减少网络传输量、降低网络I/O，从而提高整体的性能。

connections.max.idle.ms

这个参数用来指定在多久之后关闭限制的连接，默认值是540000（ms）​，即9分钟。

linger.ms

这个参数用来指定生产者发送 ProducerBatch 之前等待更多消息（ProducerRecord）加入ProducerBatch 的时间，默认值为 0。生产者客户端会在 ProducerBatch 被填满或等待时间超过linger.ms 值时发送出去。增大这个参数的值会增加消息的延迟，但是同时能提升一定的吞吐量。这个linger.ms参数与TCP协议中的Nagle算法有异曲同工之妙。

receive.buffer.bytes

这个参数用来设置Socket接收消息缓冲区（SO_RECBUF）的大小，默认值为32768（B）​，即32KB。如果设置为-1，则使用操作系统的默认值。如果Producer与Kafka处于不同的机房，则可以适地调大这个参数值。

send.buffer.bytes

这个参数用来设置Socket发送消息缓冲区（SO_SNDBUF）的大小，默认值为131072（B）​，即128KB。与receive.buffer.bytes参数一样，如果设置为-1，则使用操作系统的默认值。

request.timeout.ms

这个参数用来配置Producer等待请求响应的最长时间，默认值为30000（ms）​。请求超时之后可以选择进行重试。注意这个参数需要比broker端参数replica.lag.time.max.ms的值要大，这样可以减少因客户端重试而引起的消息重复的概率。

消费者

消费者与消费组

消费者（Consumer）负责订阅Kafka中的主题（Topic）​，并且从订阅的主题上拉取消息。与其他一些消息中间件不同的是：在Kafka的消费理念中还有一层消费组（Consumer Group）的概念，每个消费者都有一个对应的消费组。当消息发布到主题后，只会被投递给订阅它的每个消费组中的一个消费者。每个消费者只能消费所分配到的分区中的消息。换言之，每一个分区只能被一个消费组中的一个消费者所消费。

消费者与消费组这种模型可以让整体的消费能力具备横向伸缩性，我们可以增加（或减少）消费者的个数来提高（或降低）整体的消费能力。对于分区数固定的情况，一味地增加消费者并不会让消费能力一直得到提升，如果消费者过多，出现了消费者的个数大于分区个数的情况，就会有消费者分配不到任何分区。（感觉这种设计纯粹是为了保证分区消息的有序性）。以上分配逻辑都是基于默认的分区分配策略进行分析的，可以通过消费者客户端参数partition.assignment.strategy 来设置消费者与订阅主题之间的分区分配策略。

消费组是一个逻辑上的概念，它将旗下的消费者归为一类，每一个消费者只隶属于一个消费组。每一个消费组都会有一个固定的名称，消费者在进行消费前需要指定其所属消费组的名称，这个可以通过消费者客户端参数group.id来配置，默认值为空字符串。消费者并非逻辑上的概念，它是实际的应用实例，它可以是一个线程，也可以是一个进程。同一个消费组内的消费者既可以部署在同一台机器上，也可以部署在不同的机器上。

客户端开发

必要的参数配置

在Kafka消费者客户端KafkaConsumer中有4个参数是必填的：

• bootstrap.servers：该参数的释义和生产者客户端KafkaProducer 中的相同，用来 指 定 连 接 Kafka 集 群 所 需 的broker 地 址 清 单，具 体 内 容 形 式 为host1：port1，host2：post，可以设置一个或多个地址，中间用逗号隔开，此参数的默认值为“​”​。注意这里并非需要设置集群中全部的broker地址，消费者会从现有的配置中查找到全部的Kafka集群成员。这里设置两个以上的broker地址信息，当其中任意一个宕机时，消费者仍然可以连接到Kafka集群上
• group.id：消费者隶属的消费组的名称，默认值为“​”​。如果设置为空，则会报出异常：Exception in thread "main"org.apache.kafka.common.errors.InvalidGroupIdException：The configured groupId is invalid。一般而言，这个参数需要设置成具有一定的业务意义的名称。
• key.deserializer 和 value.deserializer：与生产者客户端KafkaProducer中的key.serializer和value.serializer参数对应。消费者从broker端获取的消息格式都是字节数组（byte[​]​）类型，所以需要执行相应的反序列化操作才能还原成原有的对象格式。这两个参数分别用来指定消息中key和value所需反序列化操作的反序列化器，这两个参数无默认值。注意这里必须填写反序列化器类的全限定名，比如示例中的org.apache.kafka.common.serialization.StringDeserializer，单单指定StringDeserializer是错误的

订阅主题与分区

在创建好消费者之后，我们就需要为该消费者订阅相关的主题了。一个消费者可以订阅一个或多个主题，我们可以使用subscribe（​）方法订阅了一个主题，对于这个方法而言，既可以以集合的形式订阅多个主题，也可以以正则表达式的形式订阅特定模式的主题。

对于消费者使用集合的方式（subscribe（Collection）​）来订阅主题而言，比较容易理解，订阅了什么主题就消费什么主题中的消息。如果前后两次订阅了不同的主题，那么消费者以最后一次的为准。

如果消费者采用的是正则表达式的方式（subscribe（Pattern）​）订阅，在之后的过程中，如果有人又创建了新的主题，并且主题的名字与正则表达式相匹配，那么这个消费者就可以消费到新添加的主题中的消息。如果应用程序需要消费多个主题，并且可以处理不同的类型，那么这种订阅方式就很有效。在Kafka 和其他系统之间进行数据复制时，这种正则表达式的方式就显得很常见。（可以进行正则名称匹配）

消费者不仅可以通过KafkaConsumer.subscribe（​）方法订阅主题，还可以直接订阅某些主题的特定分区，在KafkaConsumer中还提供了一个assign（​）方法来实现这些功能。这个方法只接受一个参数partitions，用来指定需要订阅的分区集合。这里补充说明一下TopicPartition类，在Kafka的客户端中，它用来表示分区，这个类的部分内容如下所示：

TopicPartition类只有2个属性：topic和partition，分别代表分区所属的主题和自身的分区编号，这个类可以和我们通常所说的主题—分区的概念映射起来。KafkaConsumer 中的partitionsFor（​）方法可以用来查询指定主题的元数据信息。其中PartitionInfo类型即为主题的分区元数据信息，此类的主要结构如下：

PartitionInfo类中的属性topic表示主题名称，partition代表分区编号，leader代表分区的leader副本所在的位置，replicas代表分区的AR集合，inSyncReplicas代表分区的ISR集合，offlineReplicas代表分区的OSR集合。

既然有订阅，那么就有取消订阅，可以使用 KafkaConsumer 中的unsubscribe（​）方法来取消主题的订阅。这个方法既可以取消通过 subscribe（Collection）方式实现的订阅，也可以取消通过subscribe（Pattern）方式实现的订阅，还可以取消通过 assign（Collection）方式实现的订阅。注意，如果将subscribe（Collection）或assign（Collection）中的集合参数设置为空集合，那么作用等同于unsubscribe（​）方法，因为新的订阅会覆盖旧的。

通过 subscribe（​）方法订阅主题具有消费者自动再均衡的功能，在多个消费者的情况下可以根据分区分配策略来自动分配各个消费者与分区的关系。当消费组内的消费者增加或减少时，分区分配关系会自动调整，以实现消费负载均衡及故障自动转移。而通过assign（​）方法订阅分区时，是不具备消费者自动均衡的功能的，其实这一点从assign（​）方法的参数中就可以看出端倪，两种类型的subscribe（​）都有ConsumerRebalanceListener类型参数的方法，而assign（​）方法却没有。

消息消费

Kafka中的消费是基于拉模式的。消息的消费一般有两种模式：推模式和拉模式。推模式是服务端主动将消息推送给消费者，而拉模式是消费者主动向服务端发起请求来拉取消息。

Kafka中的消息消费是一个不断轮询的过程，消费者所要做的就是重复地调用poll（​）方法，而poll（​）方法返回的是所订阅的主题（分区）上的一组消息。对于poll（​）方法而言，如果某些分区中没有可供消费的消息，那么此分区对应的消息拉取的结果就为空；如果订阅的所有分区中都没有可供消费的消息，那么poll（​）方法返回为空的消息集合。

注意到poll（​）方法里还有一个超时时间参数timeout，用来控制poll（​）方法的阻塞时间，在消费者的缓冲区里没有可用数据时会发生阻塞。注意这里 timeout 的类型是 Duration，它是JDK8中新增的一个与时间有关的类型。timeout的设置取决于应用程序对响应速度的要求，比如需要在多长时间内将控制权移交给执行轮询的应用线程。可以直接将timeout设置为0，这样poll（​）方法会立刻返回，而不管是否已经拉取到了消息。如果应用线程唯一的工作就是从Kafka中拉取并消费消息，则可以将这个参数设置为最大值Long.MAX_VALUE。

消费者消费到的每条消息的类型为ConsumerRecord（注意与ConsumerRecords的区别）​，这个和生产者发送的消息类型ProducerRecord相对应，不过ConsumerRecord中的内容更加丰富，具体的结构参考如下代码：

opic 和 partition 这两个字段分别代表消息所属主题的名称和所在分区的编号。offset 表示消息在所属分区的偏移量。timestamp 表示时间戳，与此对应的timestampType 表示时间戳的类型。timestampType 有两种类型：CreateTime 和LogAppendTime，分别代表消息创建的时间戳和消息追加到日志的时间戳。headers表示消息的头部内容。key 和 value 分别表示消息的键和消息的值，一般业务应用要读取的就是value，checksum是CRC32的校验值。

poll（​）方法的返回值类型是 ConsumerRecords，它用来表示一次拉取操作所获得的消息集，内部包含了若干ConsumerRecord，它提供了一个iterator（​）方法来循环遍历消息集内部的消息。

ConsumerRecords类提供了一个records（TopicPartition）方法来获取消息集中指定分区的消息。ConsumerRecords 类中并没提供与 partitions（​）类似的 topics（​）方法来查看拉取的消息集中所包含的主题列表，如果要按照主题维度来进行消费，那么只能根据消费者订阅主题时的列表来进行逻辑处理了。

到目前为止，可以简单地认为poll（​）方法只是拉取一下消息而已，但就其内部逻辑而言并不简单，它涉及消费位移、消费者协调器、组协调器、消费者的选举、分区分配的分发、再均衡的逻辑、心跳等内容，在后面的章节中会循序渐进地介绍这些内容。

位移提交

对于Kafka中的分区而言，它的每条消息都有唯一的offset，用来表示消息在分区中对应的位置。对于消费者而言，它也有一个offset的概念，消费者使用offset来表示消费到分区中某个消息所在的位置。单词“offset”可以翻译为“偏移量”​，也可以翻译为“位移”​，读者可能并没有过多地在意这一点：在很多中文资料中都会交叉使用“偏移量”和“位移”这两个词，并没有很严谨地进行区分。笔者对offset做了一些区分：对于消息在分区中的位置，我们将offset称为“偏移量”​；对于消费者消费到的位置，将offset 称为“位移”​，有时候也会更明确地称之为“消费位移”​。

在每次调用poll（​）方法时，它返回的是还没有被消费过的消息集（当然这个前提是消息已经存储在Kafka 中了，并且暂不考虑异常情况的发生）​，要做到这一点，就需要记录上一次消费时的消费位移。并且这个消费位移必须做持久化保存，而不是单单保存在内存中，否则消费者重启之后就无法知晓之前的消费位移。再考虑一种情况，当有新的消费者加入时，那么必然会有再均衡的动作，对于同一分区而言，它可能在再均衡动作之后分配给新的消费者，如果不持久化保存消费位移，那么这个新的消费者也无法知晓之前的消费位移。

在新消费者客户端中，消费位移存储在Kafka内部的主题__consumer_offsets中。这里把将消费位移存储起来（持久化）的动作称为“提交”​，消费者在消费完消息之后需要执行消费位移的提交。不过需要非常明确的是，当前消费者需要提交的消费位移并不是x，而是 x+1。读者可能看过一些相关资料，里面所讲述的内容可能是提交的消费位移就是当前所消费到的消费位移，即提交的是x，这明显是错误的。类似的错误还体现在对LEO（Log End Offset）的解读上。

KafkaConsumer 类提供了 position（TopicPartition）和committed（TopicPartition）两个方法来分别获取上面所说的position和committed offset的值。