kafka笔记4

应用程序使用KafkaConsumer向Kafka订阅主题，并从订阅的主题上接收消息。Kafka消费者从属于消费者群组，一个群组里的消费者订阅的是同一个主题，每一个消费者接收主题的一部分分区的消息。

一个分区不能被一个消费者群组里的多个消费者消费，所以若是消费者超过主题的分区数量，那么就有一部分消费者被闲置。

分区的全部权从一个消费者转移到另外一个消费者，这样的行为叫作在均衡，不过在均衡期间消费者没法读取消息，形成整个群组一小段时间不可用。

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

在读取消息以前，首先建立一个KafkaConsumer对象，有三个必选属性：bootstrap.servers,key.deserializer,value.deserializer,第四个属性group.id不是必须的，它指定了消费者属于哪一个消费者群组。

订阅主题consumer.subscribe()方法，能够指定特定主题，或使用正则表达式。消息轮询是消费者API的核心，经过一个简单的轮询向服务器请求数据。一旦消费者订阅了主题，轮询就会处理全部细节，包括群组协调，分区再均衡，发送心跳和获取数据。

返回的每条数据都包含记录所属主题信息，记录所作分区信息，记录在分区的偏移量，以及记录键值对。

在退出以前使用consume.close()关闭消费者，网络链接和socket也会随之关闭。

咱们没法让一个线程运行多个消费者，也没法让多个线程安全共享一个消费者。按照规则，一个消费者使用一个线程。

消费者的配置

1.fetch.min.bytes

该属性指定了消费者从服务器获取记录的最小字节数。

2.fetch.max.wait.ms

指定broker的等待时间，默认是500ms,这个条件和上一个条件哪个先知足，都会触发broker向消费者发送数据。

3.max.partition.fetch.bytes

该属性指定了服务器从每一个分区返回给消费者的最大字节数，默认是1MB.这个数值必须比max.message.size大。

4.session.timeout.ms

该属性指定了消费者在被认为死亡以前能够与服务器断开链接的时间，默认是3秒，若是消费者没有在这个指定时间内发送心跳给群组协调器，就会被认为已经死亡。协调器就会触发在平衡，把它的分区分配给群组的其余消费者。这个属性与heartbeat.interval.ms紧密相关，这个属性指定了消费者能够多久不发送心跳。通常同时修改这两个属性，heartbeat.interval.ms通常是session.timeout.ms的三分之一。

5.auto.offset.reset

该属性指定了消费者在读取一个没有偏移量的分区，或偏移量无效状况下该做何处理，默认值是latest，意思是偏移量无效状况下，消费者从最新记录开始读取数据，另外一个值是earliest，意思是从起始位置读取数据。

6.enable.auto.commit

该属性指定了消费者是否自动提交偏移量，默认是true，为了尽可能避免重复数据和数据丢失，能够把它设为false，由本身控制什么时候提交偏移量。

7.partition.assignment.strategy

分区会被分配给群组的消费者，partitionAssignor根据给定消费者和主题，决定哪些分区应该被分配给哪一个消费者，有两个默认分配策略：

Range:若干连续分区分配

RoundRobin:逐个分配给消费者

默认是org.apache.kafka.clients.consumer.RangeAssignor,这个类实现了Range策略，也能够改成org.apache.kafka.clients.consumer.RoundRobinAssignor

8.client.id

能够是任意字符串，broker用它标记从客户端发送过来的消息，一般被用在日志，度量指标和配额里。

9.max.poll.records

该属性用于控制单次调用call方法可以返回的记录数量

10.receive.buffer.bytes和send.buffer.bytes

socket读写数据时tcp缓冲区的大小，若是=-1，就使用操做系统默认值。