3、Kafka生产者-向Kafka写入数据

发送消息的主要步骤

格式：每个消息是一个 ProducerRecord 对象， 必须指定 所属的 Topic和Value ， 还可以指定Partition及Key

1：序列化 ProducerRecord

2：分区： 如指定Partition，不做任何事情；否则，Partitioner 根据key得到Partition 。生产者向哪个Partition发送

3：消息添加到相应 bach中 ，独立线程将batch 发到Broker上

4：broker收到消息响应 。 成功回RecordMetaData对象 ，包含了Topic信息、Patition信息、消息在Partition中的Offset信息； 失败返回错误

有序场景：不建议retries  0。可maxinflightrequestsperconnection  1， 影响生产者吞吐量，但保证有序          ps: 同partition消息有序

三个 必选 的属性：

（1） bootstrapservers ，broker地址清单

（2） keyserializer： 实现orgapachekafkacommonserializationSerializer接口的类，key序列化成字节数组。注意： 必须被设置，即使没指定key

（3）valueserializer， value序列化成字节数组

同步发送消息

异步发送消息

（1）acks： 指定多少partition副本收到消息，生产者才会认为写成功

        0，不需等待服务器的响应，吞吐量高，如broker没有收到，生产者不知道

        1，leader partition收到消息，一个即成功

        all，所有partition都收到，才成功，leader和follower共同应答

（2）buffermemory， 生产者内 缓存区域大小

（3）compressiontype ，默认不压缩，设置成snappy、gzip或lz4对发送给broker压缩

（4）retries， 重发消息的次数

（5）batchsize， 发送同一partition消息会先存储在batch中，该参数指定一个batch内存大小，单位byte。不一定填满才发送

（6）lingerms ，批次时间，batch被填满或者lingerms达到上限，就把batch中的消息发送出去

（7）maxinflightrequestsperconnection， 生产者在收到服务器响应之前可以发送的消息个数

创建ProducerRecord时，必须 指定序列化器 ，推荐序列化框架Avro、Thrift、ProtoBuf等

用 Avro 之前，先定义schema(通常用 JSON 写)

（1）创建一个类代表客户，作为消息的value

（2）定义schema

（3）生成Avro对象发送到Kafka

ProducerRecord包含Topic、value，key默认null，ey的两个作用：1）附加信息    2）被写到Topic的哪个partition

key  null ，默认partitioner， RoundRobin均衡分布

key不空，hash进行散列 ，不改变partition数量（永远不加），key和partition映射不变。

自定义paritioner 需实现Partitioner接口

（一）消费者和消费者组

1、消费者：订阅并消费kafka消息，从属于消费者组

2、消费者组：一个群组里的消费者订阅的是同一个主题，每个消费者接受主题一部分分区的消息。

注：同一个消费者可以消费不同的partition，但是同一个partition不能被不同消费者消费。

（二）消费者群组和分区再均衡

1、再均衡：分区的消费所有权从一个消费者转移到另一个消费者称为再均衡，为消费者组带来了高可用性和可伸缩性。

注：分区何时重新分配：加入消费者或者消费者崩溃等

2、如何判断消费者崩溃：消费者通过向群组协调器（某broker，不同群组可以有不同的群组协调器）发送心跳（一般在拉取消息或者提交偏移量的时候）表示自己仍旧存活，如果长时间不发送心跳则协调器认为期死亡并进行再均衡。

注：在0101版本中，心跳行为不再和获取消息和提交偏移量绑定在一起，有一个单独的心跳线程。

3、分配分区：消费者加入消费者组是，会像群组协调器发送请求，第一个加入的成为“群主”。群主从协调器那里获取成员列表，并负责给每一个消费者分配分区。完毕之后，将分配结果发送给协调器，协调器再将消息发送给所有的消费者，每个消费者只能看到自己的分配信息。只有群主知道所有的消费信息。

（三）参数配置

1、bootstrapserver：host:port

2、keyserializer：键序列化器

3、valueserializer：值序列化器

注：以上为必须设置的

4、groupid：从属的消费者组

5、fetchminbytes：消费者从服务器获取记录的最小字节数。

6、fetchmaxwaitms：消费者等待消费消息的最大时间

7、maxpartitionfetchbytes：服务器从每个分区返回给消费者的最大字节数（需要比broker的设置maxmessagesize属性配置大，否则有些消息无法消费）

8、sessiontimeoutms：指定该消费者在被认为死亡之前可以与服务器断开连接的时间，默认3秒

9、heartbeatintervalms：制定了poll方法向协调器发送心跳的频率。

注：一般9是8的三分之一

10、autooffsetreset：消费者在读取一个没有偏移量分区或者无效偏移量分区的情况下如何处理（latest：从最新记录开始读取，earliest：从最早的记录开始读取）

11、enableauthcommit：消费者是否自动提交偏移量，默认为true

12、autocommitintervalms：自动提交偏移量的时间间隔

13、partitionassignmentstrategy：分区分配给消费者的策略：

（1）range：会把主题若干个连续分区分配给消费者

（2）roundRobin：会把主题的所有分区逐个分配给消费者

14、clientid：任意字符串，broker用来区分客户端发来的消息

15：maxpollrecords：控制poll方法返回的最大记录数

16：receivebufferbytes/sendbufferbytes：tcp缓冲池读写大小

（四）订阅主题

consumersubscribe(list)

（五）轮训（消费者API的核心）

1、轮训作用： 只要消费者订阅了主题，轮训就会处理所有的细节（群组协调、分区再均衡、发送心跳、获取数据）

（1）获取数据

（2）第一次执行poll时，负责查找协调器，然后加入群组，接受分配的分区

（3）心跳的发送

（4）再均衡也是在轮训期间进行的

2、方法：poll()，消费者缓冲区没有数据时会发生阻塞，可以传一个阻塞时间，避免无限等待。0表示立即返回。

3、关闭：close()，网络连接随之关闭，立即触发再均衡。

4、线程安全：无法让一个线程运行多个消费者，也无法让多个线程公用一个消费者。

（六）提交和偏移量

1、提交：更新分区当前位置的 *** 作

2、如何提交：消费者往一个特殊主题（_consumer_offset）发送消息，消息中包含每个分区中的偏移量。

3、偏移量：分区数据被消费的位置。

4、偏移量作用：当发生再均衡时，消费者可能会分配到不一样的分区，为了继续工作，消费者需要读取到每个分区最后一次提交的偏移量，然后从偏移量的地方继续处理。

5、提交偏移量的方式

（1）自动提交：经过一个时间间隔，提交上一次poll方法返回的偏移量。每次轮训都会检测是否应该提交偏移量。缺陷：可能导致重复消费

（2）手动提交：commitSysn（）提交迁移量，最简单也最可靠，提交由poll方法返回的最新偏移量。缺点：忘了提交可能会丢数据，再均衡可能会重复消费

（3）异步提交：同步提交在提交过程中必须阻塞

（4）同步异步提交组合

（5）提交特定的偏移量

（七）再均衡监听器

（八）从特定偏移量读取数据（seek）

1、从分区开始：seekToBegining

2、从分区结束：seekToEnd

3、ConsumerRebalanceListener和seek结合使用

（九）如何退出

1、前言：wakeup方法是唯一安全退出轮训的方法，从poll方法中退出并抛出wakeupException异常。如果没有碰上轮训，则在下一次poll调用时抛出。

2、退出轮训

（1）另一个线程调用consumerwakeup方法

（2）如果循环在主线程里可以在ShutdownHook里面调用该方法

3、退出之前调用close方法：告知协调器自己要离开，出发再均衡，不必等到超时。

（十）独立消费者（assign为自己分配分区）

  Kafka是一个由Scala和Java编写的企业级的消息发布和订阅系统，最早是由Linkedin公司开发，最终开源到Apache软件基金会的项目。Kafka是一个分布式的，支持分区的，多副本的和多订阅者的高吞吐量的消息系统，被广泛应用在应用解耦、异步处理、限流削峰和消息驱动等场景。本文将针对Kafka的架构和相关组件进行简单的介绍。在介绍Kafka的架构之前，我们先了解一下Kafk的核心概念。

  在详细介绍Kafka的架构和基本组件之前，需要先了解一下Kafka的一些核心概念。

Producer： 消息的生产者，负责往Kafka集群中发送消息；

Consumer： 消息的消费者，主动从Kafka集群中拉取消息。

Consumer Group： 每个Consumer属于一个特定的Consumer Group，新建Consumer的时候需要指定对应的Consumer Group ID。

Broker： Kafka集群中的服务实例，也称之为节点，每个Kafka集群包含一个或者多个Broker（一个Broker就是一个服务器或节点）。

Message： 通过Kafka集群进行传递的对象实体，存储需要传送的信息。

Topic： 消息的类别，主要用于对消息进行逻辑上的区分，每条发送到Kafka集群的消息都需要有一个指定的Topic，消费者根据Topic对指定的消息进行消费。

Partition： 消息的分区，Partition是一个物理上的概念，相当于一个文件夹，Kafka会为每个topic的每个分区创建一个文件夹，一个Topic的消息会存储在一个或者多个Partition中。

Segment： 一个partition当中存在多个segment文件段（分段存储），每个Segment分为两部分，log文件和 index 文件，其中 index 文件是索引文件，主要用于快速查询log 文件当中数据的偏移量位置；

log文件： 存放Message的数据文件，在Kafka中把数据文件就叫做日志文件。一个分区下面默认有n多个log文件（分段存储）。一个log文件大默认1G，消息会不断追加在log文件中，当log文件的大小超过1G的时候，会自动新建一个新的log文件。

index文件： 存放log文件的索引数据，每个index文件有一个对应同名的log文件。

  后面我们会对上面的一些核心概念进行更深入的介绍。在介绍完Kafka的核心概念之后，我们来看一下Kafka的对外提供的基本功能，组件及架构设计。

  如上图所示，Kafka主要包含四个主要的API组件：

1 Producer API

  应用程序通过Producer API向Kafka集群发送一个或多个Topic的消息。

2 Consumer API

  应用程序通过Consumer API，向Kafka集群订阅一个或多个Topic的消息，并处理这些Topic下接收到的消息。

3 Streams API

  应用程序通过使用Streams API充当流处理器（Stream Processor），从一个或者多个Topic获取输入流，并生产一个输出流到一个或者多个Topic，能够有效地将输入流进行转变后变成输出流输出到Kafka集群。

4 Connect API

  允许应用程序通过Connect API构建和运行可重用的生产者或者消费者，能够把kafka主题连接到现有的应用程序或数据系统。Connect实际上就做了两件事情：使用Source Connector从数据源（如：DB）中读取数据写入到Topic中，然后再通过Sink Connector读取Topic中的数据输出到另一端（如：DB），以实现消息数据在外部存储和Kafka集群之间的传输。

  接下来我们将从Kafka的架构出发，重点介绍Kafka的主要组件及实现原理。Kafka支持消息持久化，消费端是通过主动拉取消息进行消息消费的，订阅状态和订阅关系由客户端负责维护，消息消费完后不会立刻删除，会保留历史消息，一般默认保留7天，因此可以通过在支持多订阅者时，消息无需复制多分，只需要存储一份就可以。下面将详细介绍每个组件的实现原理。

1 Producer

  Producer是Kafka中的消息生产者，主要用于生产带有特定Topic的消息，生产者生产的消息通过Topic进行归类，保存在Kafka 集群的Broker上，具体的是保存在指定的partition 的目录下，以Segment的方式（log文件和index文件）进行存储。

2 Consumer

  Consumer是Kafka中的消费者，主要用于消费指定Topic的消息，Consumer是通过主动拉取的方式从Kafka集群中消费消息，消费者一定属于某一个特定的消费组。

3 Topic

  Kafka中的消息是根据Topic进行分类的，Topic是支持多订阅的，一个Topic可以有多个不同的订阅消息的消费者。Kafka集群Topic的数量没有限制，同一个Topic的数据会被划分在同一个目录下，一个Topic可以包含1至多个分区，所有分区的消息加在一起就是一个Topic的所有消息。

4 Partition

  在Kafka中，为了提升消息的消费速度，可以为每个Topic分配多个Partition，这也是就之前我们说到的，Kafka是支持多分区的。默认情况下，一个Topic的消息只存放在一个分区中。Topic的所有分区的消息合并起来，就是一个Topic下的所有消息。每个分区都有一个从0开始的编号，每个分区内的数据都是有序的，但是不同分区直接的数据是不能保证有序的，因为不同的分区需要不同的Consumer去消费，每个Partition只能分配一个Consumer，但是一个Consumer可以同时一个Topic的多个Partition。

5 Consumer Group

  Kafka中的每一个Consumer都归属于一个特定的Consumer Group，如果不指定，那么所有的Consumer都属于同一个默认的Consumer Group。Consumer Group由一个或多个Consumer组成，同一个Consumer Group中的Consumer对同一条消息只消费一次。每个Consumer Group都有一个唯一的ID，即Group ID，也称之为Group Name。Consumer Group内的所有Consumer协调在一起订阅一个Topic的所有Partition，且每个Partition只能由一个Consuemr Group中的一个Consumer进行消费，但是可以由不同的Consumer Group中的一个Consumer进行消费。如下图所示：

在层级关系上来说Consumer Group好比是跟Topic对应的，而Consumer就对应于Topic下的Partition。Consumer Group中的Consumer数量和Topic下的Partition数量共同决定了消息消费的并发量，且Partition数量决定了最终并发量，因为一个Partition只能由一个Consumer进行消费。当一个Consumer Group中Consumer数量超过订阅的Topic下的Partition数量时，Kafka会为每个Partition分配一个Consumer，多出来的Consumer会处于空闲状态。当Consumer Group中Consumer数量少于当前定于的Topic中的Partition数量是，单个Consumer将承担多个Partition的消费工作。如上图所示，Consumer Group B中的每个Consumer需要消费两个Partition中的数据，而Consumer Group C中会多出来一个空闲的Consumer4。总结下来就是：同一个Topic下的Partition数量越多，同一时间可以有越多的Consumer进行消费，消费的速度就会越快，吞吐量就越高。同时，Consumer Group中的Consumer数量需要控制为小于等于Partition数量，且最好是整数倍：如1，2，4等。

6 Segment

  考虑到消息消费的性能，Kafka中的消息在每个Partition中是以分段的形式进行存储的，即每1G消息新建一个Segment，每个Segment包含两个文件：log文件和index文件。之前我们已经说过，log文件就是Kafka实际存储Producer生产的消息，而index文件采用稀疏索引的方式存储log文件中对应消息的逻辑编号和物理偏移地址（offset），以便于加快数据的查询速度。log文件和index文件是一一对应，成对出现的。下图展示了log文件和index文件在Partition中的存在方式。

  Kafka里面每一条消息都有自己的逻辑offset（相对偏移量）以及存在物理磁盘上面实际的物理地址便宜量Position，也就是说在Kafka中一条消息有两个位置：offset（相对偏移量）和position（磁盘物理偏移地址）。在kafka的设计中，将消息的offset作为了Segment文件名的一部分。Segment文件命名规则为：Partition全局的第一个Segment从0开始，后续每个segment文件名为上一个Partition的最大offset（Message的offset，非实际物理地偏移地址，实际物理地址需映射到log中，后面会详细介绍在log文件中查询消息的原理）。数值最大为64位long大小，由20位数字表示，前置用0填充。

  上图展示了index文件和log文件直接的映射关系，通过上图，我们可以简单介绍一下Kafka在Segment中查找Message的过程：

   1 根据需要消费的下一个消息的offset，这里假设是7，使用二分查找在Partition中查找到文件名小于（一定要小于，因为文件名编号等于当前offset的文件里存的都是大于当前offset的消息）当前offset的最大编号的index文件，这里自然是查找到了00000000000000000000index。

   2 在index文件中，使用二分查找，找到offset小于或者等于指定offset（这里假设是7）的最大的offset，这里查到的是6，然后获取到index文件中offset为6指向的Position（物理偏移地址）为258。

   3 在log文件中，从磁盘位置258开始顺序扫描，直到找到offset为7的Message。

至此，我们就简单介绍完了Segment的基本组件index文件和log文件的存储和查询原理。但是我们会发现一个问题：index文件中的offset并不是按顺序连续存储的，为什么Kafka要将索引文件设计成这种不连续的样子？这种不连续的索引设计方式称之为稀疏索引，Kafka中采用了稀疏索引的方式读取索引，kafka每当log中写入了4k大小的数据，就往index里以追加的写入一条索引记录。使用稀疏索引主要有以下原因：

   (1) 索引稀疏存储，可以大幅降低index文件占用存储空间大小。

   (2) 稀疏索引文件较小，可以全部读取到内存中，可以避免读取索引的时候进行频繁的IO磁盘 *** 作，以便通过索引快速地定位到log文件中的Message。

7 Message

  Message是实际发送和订阅的信息是实际载体，Producer发送到Kafka集群中的每条消息，都被Kafka包装成了一个Message对象，之后再存储在磁盘中，而不是直接存储的。Message在磁盘中的物理结构如下所示。

  其中 key 和 value 存储的是实际的Message内容，长度不固定，而其他都是对Message内容的统计和描述，长度固定。因此在查找实际Message过程中，磁盘指针会根据Message的 offset 和 message length 计算移动位数，以加速Message的查找过程。之所以可以这样加速，因为Kafka的log文件都是顺序写的，往磁盘上写数据时，就是追加数据，没有随机写的 *** 作。

8Partition Replicas

  最后我们简单聊一下Kafka中的Partition Replicas（分区副本）机制，08版本以前的Kafka是没有副本机制的。创建Topic时，可以为Topic指定分区，也可以指定副本个数。kafka 中的分区副本如下图所示：

  Kafka通过副本因子（replication-factor）控制消息副本保存在几个Broker（服务器）上，一般情况下副本数等于Broker的个数，且同一个副本因子不能放在同一个Broker中。副本因子是以分区为单位且区分角色；主副本称之为Leader（任何时刻只有一个），从副本称之为 Follower（可以有多个），处于同步状态的副本叫做in-sync-replicas(ISR)。Leader负责读写数据，Follower不负责对外提供数据读写，只从Leader同步数据，消费者和生产者都是从leader读写数据，不与follower交互，因此Kafka并不是读写分离的。同时使用Leader进行读写的好处是，降低了数据同步带来的数据读取延迟，因为Follower只能从Leader同步完数据之后才能对外提供读取服务。

  如果一个分区有三个副本因子，就算其中一个挂掉，那么只会剩下的两个中，选择一个leader，如下图所示。但不会在其他的broker中，另启动一个副本（因为在另一台启动的话，必然存在数据拷贝和传输，会长时间占用网络IO，Kafka是一个高吞吐量的消息系统，这个情况不允许发生）。如果指定分区的所有副本都挂了，Consumer如果发送数据到指定分区的话，将写入不成功。Consumer发送到指定Partition的消息，会首先写入到Leader Partition中，写完后还需要把消息写入到ISR列表里面的其它分区副本中，写完之后这个消息才能提交offset。

  到这里，差不多把Kafka的架构和基本原理简单介绍完了。Kafka为了实现高吞吐量和容错，还引入了很多优秀的设计思路，如零拷贝，高并发网络设计，顺序存储，以后有时间再说。

以上就是关于3、Kafka生产者-向Kafka写入数据全部的内容，包括:3、Kafka生产者-向Kafka写入数据、consumer（KafkaConsumer）、Kafka架构及基本原理简析等相关内容解答，如果想了解更多相关内容，可以关注我们，你们的支持是我们更新的动力！