《Kafka权威指南》读书笔记

一、 Kafka的组成

• kafka是一个发布与订阅消息系统
• 消息：kafka的数据单元称为"消息"。可以把消息看成是数据库中的一个"数据行"。

消息的key：为key生成一个一致性散列值(HashCode)，然后使用散列值对主题分区数进行取模，为消息选取分区。

• 消息被分批次写入kafka。

批次：就是一组消息，这些消息属于同一个主题和分区。

• 主题(topic)：kafka的消息，是通过topic来分类的。topic，好比数据库里的表，或者文件系统里的文件夹。

topic，可分为若干个分区(partition)。

由于一个topic一般分为几个partition，因此整个Topic范围内无法保证消息的顺序，但可以保证消息在单个Partition内的顺序。

• 生产者(producer)：创建消息。

• 消费者(consumer)：订阅读取消息。

consumer可以订阅一个或多个topic，并按照消息生成的顺序去读取它们。消费者通过检查消息偏移量来区分已经读取过的消息。

• 偏移量(offset)：是一个不断递增的整数值，在创建消息时，Kafka会把它添加到消息里。在给定的分区里，每个消息的偏移量都是唯一的。

• 消费者群组： 消费者群组中，会有一个或多个消费者读取同一个topic。
  另外，群组能保证每个分区只能被一个消费者使用。

• 服务器(broker)：Kafka的服务器broker，接收生产者的消息，为消息设置偏移量，并提交消息到磁盘保存。broker为消费者提供服务，对请求作出响应，返回已经提交到磁盘上的消息。broker是集群的重要组成部分。

为什么选择kafka?

• 多个生产者.
• 多个消费者.

多个消费者，能够提高消息的处理效率。

• 基于磁盘的数据存储.
• 伸缩性。

kafka可以灵活地配置broker的个数，根据生产环境的需要进行调整。

• 高性能.

二、Kafka配置

broker配置

• broker.id：服务器id
• port：端口号
• zookeeper.connect:用于保存broker元数据的zookeeper地址是通过此配置来指定的。
• log.dirs：磁盘存放日志的路径。
• auto.create.topics.enable：是否自动创建topic。

topic配置

• num.partitions：指定topic包含多少个分区。
• log.retention.ms：指定日志保留的时间。
• log.retention.bytes：通过保留的消息字节数来判断消息是否过期。
• message.max.bytes:限制单个消息的大小。

三、kafka生产者--向Kafka写入数据

kafka生产者组件图：

图片/kafka分区与消费者的关系

• kafka生产者有3个必选属性：

bootstrap.servers：指定broker的地址清单。建议提供两个以上的broker信息，一旦其中一个宕机，生产者仍然能够连接上集群。

key.serializer：broker的消息的key和value都是字节数组，因此需要序列化。生产者得知道如何把java对象转换成字节数组。key.serializer必须一个实现了Serializer接口的类，生产者会使用这个类把键对象序列化为字节数组。

value.serializer：同上。

生产者发送消息的三种方式：

• 发送并忘记(fire-and forge)：生产者把消息发送给服务器，但并不关心它是否正常到达。生产者会自动尝试重发，不过使用这种方式有时候也会丢失一些消息。

• 同步发送：send()方法发送消息，会返回一个Future对象，调用get()方法进行等待，就可以知道消息是否发送成功。

• 异步发送：调用send()方法，并指定一个回调函数，服务器在返回响应时调用该函数。

生产者使用多线程：

生产者是可以使用多线程来发送消息的。如果需要更高的吞吐量，可以在生产者数量不变的前提下增加线程数量。如果这样做还不够，可以增加生产者数量。

生产者配置：

• key.serializer
  用于 key 键的序列化，它实现了 org.apache.kafka.common.serialization.Serializer 接口

• value.serializer
  用于 value 值的序列化，实现了 org.apache.kafka.common.serialization.Serializer 接口

• acks
  acks 参数指定了要有多少个分区副本接收消息，生产者才认为消息是写入成功的。此参数对消息丢失的影响较大

如果 acks = 0，就表示生产者也不知道自己产生的消息是否被服务器接收了，它才知道它写成功了。如果发送的途中产生了错误，生产者也不知道，它也比较懵逼，因为没有返回任何消息。这就类似于 UDP 的运输层协议，只管发，服务器接受不接受它也不关心。

如果 acks = 1，只要集群的 Leader 接收到消息，就会给生产者返回一条消息，告诉它写入成功。如果发送途中造成了网络异常或者 Leader 还没选举出来等其他情况导致消息写入失败，生产者会受到错误消息，这时候生产者往往会再次重发数据。因为消息的发送也分为 同步 和 异步，Kafka 为了保证消息的高效传输会决定是同步发送还是异步发送。如果让客户端等待服务器的响应（通过调用 Future 中的 get() 方法），显然会增加延迟，如果客户端使用回调，就会解决这个问题。

如果 acks = all，这种情况下是只有当所有参与复制的节点都收到消息时，生产者才会接收到一个来自服务器的消息。不过，它的延迟比 acks =1 时更高，因为我们要等待不只一个服务器节点接收消息。

• buffer.memory
  此参数用来设置生产者内存缓冲区的大小，生产者用它缓冲要发送到服务器的消息。如果应用程序发送消息的速度超过发送到服务器的速度，会导致生产者空间不足。这个时候，send() 方法调用要么被阻塞，要么抛出异常，具体取决于 block.on.buffer.null 参数的设置。
  compression.type
  此参数来表示生产者启用何种压缩算法，默认情况下，消息发送时不会被压缩。该参数可以设置为 snappy、gzip 和 lz4，它指定了消息发送给 broker 之前使用哪一种压缩算法进行压缩。下面是各压缩算法的对比

• retries
  生产者从服务器收到的错误有可能是临时性的错误（比如分区找不到首领），在这种情况下，reteis 参数的值决定了生产者可以重发的消息次数，如果达到这个次数，生产者会放弃重试并返回错误。默认情况下，生产者在每次重试之间等待 100ms，这个等待参数可以通过 retry.backoff.ms 进行修改。

• batch.size
  当有多个消息需要被发送到同一个分区时，生产者会把它们放在同一个批次里。该参数指定了一个批次可以使用的内存大小，按照字节数计算。当批次被填满，批次里的所有消息会被发送出去。不过生产者井不一定都会等到批次被填满才发送，任意条数的消息都可能被发送。

• client.id
  此参数可以是任意的字符串，服务器会用它来识别消息的来源，一般配置在日志里。

• max.in.flight.requests.per.connection
  此参数指定了生产者在收到服务器响应之前可以发送多少消息，它的值越高，就会占用越多的内存，不过也会提高吞吐量。把它设为1 可以保证消息是按照发送的顺序写入服务器。

• timeout.ms、request.timeout.ms 和 metadata.fetch.timeout.ms

request.timeout.ms 指定了生产者在发送数据时等待服务器返回的响应时间，metadata.fetch.timeout.ms 指定了生产者在获取元数据（比如目标分区的首领是谁）时等待服务器返回响应的时间。如果等待时间超时，生产者要么重试发送数据，要么返回一个错误。timeout.ms 指定了 broker 等待同步副本返回消息确认的时间，与 asks 的配置相匹配----如果在指定时间内没有收到同步副本的确认，那么 broker 就会返回一个错误。

• max.block.ms
  此参数指定了在调用 send() 方法或使用 partitionFor() 方法获取元数据时生产者的阻塞时间当生产者的发送缓冲区已捕，或者没有可用的元数据时，这些方法就会阻塞。在阻塞时间达到 max.block.ms 时，生产者会抛出超时异常。

• max.request.size
  该参数用于控制生产者发送的请求大小。它可以指能发送的单个消息的最大值，也可以指单个请求里所有消息的总大小。

• receive.buffer.bytes 和 send.buffer.bytes
  Kafka 是基于 TCP实现的，为了保证可靠的消息传输，这两个参数分别指定了 TCP Socket 接收和发送数据包的缓冲区的大小。如果它们被设置为 -1，就使用操作系统的默认值。如果生产者或消费者与 broker 处于不同的数据中心，那么可以适当增大这些值。

生产者发送消息的顺序保证

• kafka可以保证同一个分区里的消息是有序的。也就是说生产者按照一定的顺序发送消息，broker就会按照这个顺序把它们写入分区，消费者也会按照顺序读取它们。

• 如果把retries设为非零整数，同时把max.in.flight.requests.per.connection设为比1大的值，那么如果第一个批次消息写入失败，而第二个批次写入成功，broker会重试写入第一个批次。如果此时第一个批次也写入成功，那么两个批次的顺序就反过来了。
  可以把max.in.flight.requests.per.connection设为1，这样在生产者尝试发送第一批消息时，就不会有其他的消息发送给broker。对消息的顺序有严格要求的情况下才能这么做，不然会影响吞吐量。

四、Kafka消费者--从Kafka读取消息

分区和消费者的关系

• 为什么kafka每个Partition，只能被消费者群组中的一个消费者消费？

假设群组内的多个消费者负责同一个分区，那么会有什么问题呢？

我们知道，Kafka它在设计的时候就是要保证分区下消息的顺序，也就是说消息在一个分区中的顺序是怎样的，那么消费者在消费的时候看到的就是什么样的顺序，那么要做到这一点就首先要保证消息是由消费者主动拉取的（pull），其次还要保证一个分区只能由一个消费者负责。倘若，两个消费者负责同一个分区，那么就意味着两个消费者同时读取分区的消息，由于消费者自己可以控制读取消息的offset，就有可能C1才读到2，而C1读到1，C1还没处理完，C2已经读到3了，则会造成很多浪费，因为这就相当于多线程读取同一个消息，会造成消息处理的重复，且不能保证消息的顺序，这就跟主动推送（push）无异。

• 假如两个不同群组的消费者，分别去消费同一个分区，那么分区消息上的偏移量会怎么变化？这两个消费者，最终能读取到相同的消息么？

不同的消费者群组可以从同一主题获取所有的消息，消费者群组G1和消费者群组G2之间互不影响。

多个应用程序可以从同一主题获取到所有的消息。

只要保证每个订阅的应用程序都有自己的不同的消费者群组，每个订阅的应用程序都可以从同一主题获取所有的消息，而不只是其中的一部分。

轮询

轮询是消费者api的核心。通过一个简单的轮询向服务器请求数据，一旦消费者订阅了主题，轮询就会处理群组协调、分区再均衡、发送心跳、获取数据。

其他

• 内存映射

概括：用户空间的一段内存区域映射到内核空间，这样，无论是内核空间或用户空间对这段内存区域的修改，都可以直接映射到另一个区域。
优势：如果内核态和用户态存在大量的数据传输，效率是非常高的。
为什么会提高效率：概括来讲，传统方式为read()系统调用，进行了两次数据拷贝；内存映射方式为mmap()系统调用，只进行一次数据拷贝

• 零拷贝方式：
  如何让数据不经过用户空间？零拷贝省略了拷贝到用户缓冲的步骤，通过文件描述符，直接从内核空间将数据复制到网卡接口。