Kafka:消费者详解
Posted Icedzzz
tags:
篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了Kafka:消费者详解相关的知识，希望对你有一定的参考价值。
文章目录
1. Kafka消费者概念
消费者： 负责订阅Kafka的topic,并从订阅的topic中拉取消息并验证，最后将他们保存起来。
消费者组： 如果只使用一个消费者处理消息，那应用程序会远跟不上消息生产的速度。因此引入消费者组这个概念，对消费者进行横向伸缩，当消息发布到topic后，会被投递给订阅它的每个消费者组中的一个消费者。
注意：
- 一个消费者组里面的消费者订阅的是同一个主题，每个消费者接收主题的一部分分区的消息。
- 消费组是逻辑上的概念，它将旗下消费者归为一类，每个消费者只属于一个消费组。
- 消费者不是逻辑上的概念，它是实际应用实例，它是一个线程/进程。
往群组里增加消费者是横向伸缩消费能力的主要方式： Kafka消费者在进行一些耗时的计算与操作的情况下，单个消费者无法跟上数据生成的速度，于是通过扩张消费者数量，同时为topic创建大量的分区，让每个消费者只处理部分分区的消息。值得注意的是，横向伸缩Kafka消费者和消费者群组不会对性能造成负面影响。
Kafka同时支持点对点和发布/订阅两种消息投递模式：
- 如果所有消费者都属于应该消费组，那么所有消息都会被均衡的投递给每个消费者，这就相对于点对点模式
- 如果消费者隶属于不同的消费组，那么所有的消息都会广播给所有的消费者，这就相对于发布/订阅模式
对于消息中间件而言，一般有两种消息投递模式：点对点（P2P)模式和发布/订阅(Pub/Sub)模式.点对点模式是基于队列的，生产者发送消息到队列，消费者从队列中接收消息。发布订阅模式定义了如何向应该内容节点发布和订阅消息，这个内容节点称为主题(Topic)。主题使消息的订阅者和发布者互相独立，不需要进行接触即可保证消息的传递，发布/订阅模式在消息的一对多广播时采用。
(1) 如果消费者组中只有一个消费者，那么这个消费者将收到对应topic所有分区的消息：
(2) 如果在(1)中新增一个消费者，那么每个消费者将分别从两个分区中接收消息：
(3) 如果消费者数量和分区数量相等，那么每个消费者可以分配到一个分区：
(4) 如果消费者数量大于分区数量，那么有一部分消费者被闲置，不会接收到任何消息，实际应用中应该尽量避免该问题：
(5) 多个消费者消费同一个topic的数据，互不影响：
(每个消费者组都会有一个固定的名称，消费者在进行消费前需要指定其所属消费组的名称，这个参数由消费组客户端参数group.id来配置)
2. Kafka客户端开发
步骤：
- 配置消费者参数及创建相应消费者实例
- 订阅topic
- 拉取消息并消费
- 提交消费位移
- 关闭消费者实例
public class KafkaConsumerAnalysis
public static final String brokerList="node01:9092";
public static final String topic="topic-demo";
public static final String gruopId="group.demo";
public static final AtomicBoolean isRunning=new AtomicBoolean(true);
public static Properties initConfig()
Properties prop = new Properties();
prop.put("bootstrap.servers",brokerList);
prop.put("group.id",gruopId);
prop.put("client.id","consumer.client.id.Demo");
prop.put("key.deserializer", "org.apache.kafka.common.serialization.StringSerializer");
prop.put("value.deserializer", "org.apache.kafka.common.serialization.StringSerializer");
return prop;
public static void main(String[] args)
Properties properties = initConfig();
KafkaConsumer<String,String> consumer = new KafkaConsumer<>(properties);
consumer.subscribe(Arrays.asList(topic));
try
while (isRunning.get())
ConsumerRecords<String, String> records = consumer.poll(1000);
for (ConsumerRecord<String, String> record : records)
System.out.println("topic="+record.topic()+
",partition="+record.partition()+
",offset="+record.offset()+
",key="+record.offset()+
",value="+record.value());
catch (Exception e)
e.printStackTrace();
finally
consumer.close();
consumer.subscribe负责对主题的订阅，Kafka支持consumer.subscribe(Collection)、consumer.subscribe(Pattern)、assign(Collection)三种方式订阅主题，分别为订阅主题集合、正则表达式订阅和指定分区订阅三种方式，这三种方式是互斥的，在消费者中只能使用一种，否则会报IllegalStateException异常。
特别注意的是： subscribe方法订阅主题具有消费者自动再均衡的功能，在多个消费者的情况下可以根据分区分配策略来自动分配各个消费者与分区之间的关系，实现消费负载均衡及故障自动转移。assign方法不具备该功能。
**consumer.poll()**是消费者应用的关键，消费者必须持续不断对Kafka进行轮询，否则就会被认为已经死亡，它的分区会被移交给其他消费者。poll中传入的参数为超时实践，用于控制poll方法的堵塞时间。consumer.poll会返回一个记录列表。每条记录都包含了记录所属主题的信息、记录所在分区的信息、记录在分区里的偏移量，以及记录的键值对。
除了上述几个配置参数，还有几个参数需要注意：
-
auto.offset.reset ：该属性指定了消费者在读取一个没有偏移量的分区或者偏移量无效的情况下（因消费者长时间失效，包含偏移量的记录已经过时并被删除）该作何处理。它的默认值是 latest，意
思是说，在偏移量无效的情况下，消费者将从最新的记录开始读取数据（在消费者启动之后生成的记录）。另一个值是 earliest，意思是说，在偏移量无效的情况下，消费者将从起始位置读取分区的记录 -
enable.auto.commit ：该属性指定了消费者是否自动提交偏移量，默认值是 true。为了尽量避免出现重复数据和数据丢失，可以把它设为 false，由自己控制何时提交偏移量。如果把它设为 true，还可以通过配置 auto.commit.interval.ms属性来控制提交的频率。
-
partition.assignment.strategy ：Kafka 有两个默认的分配策略：Range和RoundRobin。
默 认 使 用 的 是 org.apache.kafka.clients.consumer.RangeAssignor，这个类实现了 Range 策略，不过也可以把它改成 org.apache.kafka.clients.consumer.RoundRobinAssignor。
3. 深入理解消费者模式
1. 消费者消费模式
Kafka的消费者采用从broker 拉取(pull) 消息的方式获取消息，生产者采用 **推送(push）**方式推送消息给broker。push模式是服务器主动推送消息，pull模式是消费者主动向服务端发起请求拉取消息。Kafka消费消息是一个不断轮询的过程，消费者只要不断重复调用poll方法即可， poll() 方法返回由生产者写入 Kafka 但还没有被消费者读取过的记录。
poll源码：
public ConsumerRecords<K, V> poll(long timeout)
acquire();
try
if (timeout < 0)
throw new IllegalArgumentException("Timeout must not be negative");
if (this.subscriptions.hasNoSubscriptionOrUserAssignment())
throw new IllegalStateException("Consumer is not subscribed to any topics or assigned any partitions");
long start = time.milliseconds();
long remaining = timeout;
do
Map<TopicPartition, List<ConsumerRecord<K, V>>> records = pollOnce(remaining);
if (!records.isEmpty())
if (fetcher.sendFetches() > 0 || client.pendingRequestCount() > 0)
client.pollNoWakeup();
if (this.interceptors == null)
return new ConsumerRecords<>(records);
else
return this.interceptors.onConsume(new ConsumerRecords<>(records));
long elapsed = time.milliseconds() - start;
remaining = timeout - elapsed;
while (remaining > 0);
return ConsumerRecords.empty();
finally
release();
值得注意的是**acquire()和release()**的调用，KafkaProducer是线程安全的，但KafkaConsumer并不是线程安全的。
poll中调用的acquire方法是用来检测当前是否只有一个线程在操作，如果其他线程在操作则会抛出 ConcurrentModificationException(“KafkaConsumer is not safe for multi-threaded access”)异常。acquire方法中并没有调用锁(synchronized、juc的lock），它通过线程操作计数标记的方式来检测线程是否发生了并发操作，以此保证只有一个线程在操作。release方法与acquire成对出现，负责相应的“解锁”。
acquire源码：
private final AtomicInteger refcount = new AtomicInteger(0);
private void acquire()
ensureNotClosed();
long threadId = Thread.currentThread().getId();
if (threadId != currentThread.get() && !currentThread.compareAndSet(NO_CURRENT_THREAD, threadId))
throw new ConcurrentModificationException("KafkaConsumer is not safe for multi-threaded access");
refcount.incrementAndGet();
2. 偏移量问题
消费者是如何提交偏移量的呢？消费者通过往一个叫作 _consumer_offset 的特殊主题发送消息，消息里包含每个分区的偏移量。如果消费者一直处于运行状态，那么偏移量就没有什么用处。不过，如果消费者发生崩溃或者有新的消费者加入群组，就会触发****再均衡，完成再均衡之后，每个消费者可能分配到新的分区，而不是之前处理的那个。为了能够继续之前的工作，消费者需要读取每个分区最后一次提交的偏移量，然后从偏移量指定的地方继续处理。
如果提交的偏移量小于客户端处理的最后一个消息的偏移量，那么处于两个偏移量之间的消息就会被重复处理，导致数据重复：
如果提交的偏移量大于客户端处理的最后一个消息的偏移量，那么处于两个偏移量之间的消息将会丢失：
3. 四种提交偏移量方式
- 自动提交偏移量
设置 enable.auto.commit 被设为 true，即可让消费者自动提交偏移量，默认间隔时间5s。也可以通过auto.commit.interval.ms 设置。自动提交是在轮询里进行的，消费者每次在进行轮询时会检查是否该提交偏移量了，如果是，那么就会提交从上一次轮询返回的偏移量。
存在的问题： 默认5s自动提交偏移量，如果在最近一次提交之后的 3s 发生了再均衡，再均衡之后，消费者从最后一次提交的偏移量位置开始读取消息。这个时候偏移量已经落后了 3s，所以在这 3s 内到达的消息会被重复处理。
- 手动，同步提交偏移量
首先将auto.commit.offset 设为 false，设置为手动提交偏移量。commitSync() 方法可以提交由 poll() 方法返回的最新偏移量，提交成功后马上返回，如果提交失败就抛出异常，如是可修复的异常则会不断重试 。但在实际应用中，很少会有这种消费一条消息就提交一次消费偏移量的必要场景，这样会消耗大量的性能。
while (true)
ConsumerRecords<String, String> records = consumer.poll(100);
for (ConsumerRecord<String, String> record : records
System.out.printf("topic = %s, partition = %s, offset =
%d, customer = %s, country = %s\\n",
record.topic(), record.partition(),
record.offset(), record.key(), record.value());
try
consumer.commitSync();
catch (CommitFailedException e)
log.error("commit failed", e)
在实际应用中，往往需要通过按照分区粒度划分来提交偏移量，这时候需要用到ConsumerRecords提供的方法。ConsumerRecords类提供一个records(TopPar…)方法来获取消息集中指定分区消息，也有records(topic)的重载方法，提供指定主题列表，按照主题维度进行消费，同时ConsumerRecords.partitions方法用来获取消息集中所有分区。
按照分区粒度提交偏移量：
while (isRunning.get())
ConsumerRecords<String, String> records = consumer.poll(1000);
for (ConsumerRecord<String, String> record : records)
System.out.println("topic="+record.topic()+
",partition="+record.partition()+
",offset="+record.offset()+
",key="+record.offset()+
",value="+record.value());
for (TopicPartition partition : records.partitions())
List<ConsumerRecord<String, String>> partitionRecords = records.records(partition);
for (ConsumerRecord<String, String> record : partitionRecords)
//...
long lastConsumedOffset = partitionRecords.get(partitionRecords.size() - 1).offset();
consumer.commitSync(Collections.singletonMap(partition,new OffsetAndMetadata(lastConsumedOffset+1)));
注意： 当前消费位移为lastConsumedOffset，下一次需要提交的唯一为lastConsumedOffset+1，而不是lastConsumedOffset
- 手动，异步提交偏移量
自动提交有一个不足之处，在 broker 对提交请求作出回应之前，应用程序会一直阻塞，这样会限制应用的吞吐量，如果通过降低提交频率来提升吞吐量，则再发生再均衡时，会增加重复消息的数量。与commitSync方法相反，异步提交方法commitAsync() 在执行时消费者线程不会被堵塞，可能在提交偏移量的结果还未返回之前就开始了新一次的拉取操作。commitAsync共有三个构造方法，其中第二第三个构造方法在偏移量提交完成后会调用OffsetCommitCallback 的onComplete 回调方法。
public void commitAsync()
public void commitAsync(OffsetCommitCallback callback)
public void commitAsync(final Map<TopicPartition, OffsetAndMetadata> offsets, OffsetCommitCallback callback)
注意： 在成功提交或碰到无法恢复的错误之前， commitSync() 会一直重试，但是 commitAsync()不会，这也是 commitAsync() 不好的一个地方。在成功提交或碰到无法恢复的错误之前， commitSync() 会一直重试，但是 commitAsync()不会，这也是 commitAsync() 不好的一个地方。它之所以不进行重试，是因为在它收到服务器响应的时候，可能有一个更大的偏移量已经提交成功。假设我们发出一个请求用于提交偏移量 2000，这个时候发生了短暂的通信问题，服务器收不到请求，自然也不会作出任何响应。与此同时，我们处理了另外一批消息，并成功提交了偏移量 3000。如果commitAsync() 重新尝试提交偏移量 2000，它有可能在偏移量 3000 之后提交成功。这个时候如果发生再均衡，就会出现重复消息。
**为了避免这种情况发生，我们可以设置一个递增的序号来维护异步提交的顺序，**每次提交后就增加序号相应的值。在进行重试前，先检查回调的序列号和即将提交的偏移量是否相等，如果相等，说明没有新的提交，那么可以安全地进行重试。如果序列号比较大，说明有一个新的提交已经发送出去了，应该停止重试。
- 异步同步结合提交偏移量
当消费者异常退出时，由于异步提交不会进行重试，很大概率会发生重复消息。所以在消费者退出或者再均衡执行之前，使用同步提交做最后的重试，直到提交成功或发生无法恢复的错误
try
while (true)
ConsumerRecords<String, String> records = consumer.poll(100);
for (ConsumerRecord<String, String> record : records)
System.out.println("topic = %s, partition = %s, offset = %d,
customer = %s, country = %s\\n",
record.topic(), record.partition(),
record.offset(), record.key(), record.value());
consumer.commitAsync();
catch (Exception e)
log.error("Unexpected error", e);
finally
try
consumer.commitSync();
finally
consumer.close();
4. 再均衡
-
什么是再均衡问题？
当一个新的消费者加入群组时，它读取的是原本由其他消费者读取的消息。当一个消费者被关闭或发生崩溃时，它就离开群组，原本由它读取的分区将由群组里的其他消费者来读取。 这种消费者的行为导致broker分区的所有权从一个消费者转移到另外一个消费者，这种行为被称为再均衡。 -
再均衡的优缺点
再均衡可以保证消费者群组的 高可用性 和 伸缩性 ，但在正常情况下，并不希望发生再均衡。因为在在均衡期间，消费者无法从broker中读取消息，造成整个群组一段时间不可用。同时，当分区被重新分配给另外个消费者时，消费者当前的读取状态会丢失，它有可能还需要去刷新缓存，在它重新恢复状态之前会拖慢应用程序。 -
心跳机制：
消费者通过向被指派为群组协调器的 broker（不同的群组可以有不同的协调器）发送心跳来维持它们和群组的从属关系以及它们对分区的所有权关系。只要消费者以正常的时间间隔发送心跳，就被认为是活跃的，说明它还在读取分区里的消息。消费者会在轮询消息（为了获取消息）或提交偏移量时发送心跳。如果消费者停止发送心跳的时间足够长，会话就会过期，群组协调器认为它已经死亡，就会触发一次再均衡。
如果一个消费者发生崩溃，并停止读取消息，群组协调器会等待几秒钟，确认它死亡了才会触发再均衡。在这几秒钟时间里，死掉的消费者不会读取分区里的消息。在清理消费者时，消费者会通知协调器它将要离开群组，协调器会立即触发一次再均衡，尽量降低处理停顿。 -
新版本中的改动：
Kafka 社区引入了一个独立的心跳线程，可以在轮询消息的空档发送心跳。这样一来，发送心跳的频率（也就是消费者群组用于检测发生崩溃的消费者或不再发送心跳的消费者的时间）与消息轮询的频率（由处理消息所花费的时间来确定）之间就是相互独立的。在新版本的 Kafka 里，可以指定消费者在离开群组并触发再均衡之前可以有多长时间不进行消息轮询，这样可以避免出现活锁（livelock），比如有时候应用程序并没有崩溃，只是由于某些原因导致无法正常运行。这个配置与session.timeout.ms 是相互独立的，后者用于控制检测消费者发生崩溃的时间和停止发送心跳的时间
5. 再均衡监听器
再均衡监听器用来设定发生在再均衡前后的一些准备或收尾的动作，以避免出现 重复消费现象（一个消费者消费完某个分区中的一部分消息时还未来得及提交消息就发生了再均衡，之后这个分区又被分配给另外一个消费者，原来被消费完的那部分消息又重新被消费了一遍）。
通过在subscribe()方法中继承再均衡监听器ConsumerRebalanceListener接口，实现如下两个方法：
//在再均衡开始之前或者消费者停止读取消息之后调用，通常通过这个回调方法处理消费位移的提交，避免重复消费
public void onPartitionsRevoked(Collection<TopicPartition> partitions)
//重新分配分区之后和新的消费者开始读取消息之前调用
public void onPartitionsAssigned(Collection<TopicPartition> partitions)
注意 ：onPartitionsRevoked的partitions表示再均衡之前所分配的分区；onPartitionsAssigned的partitions表示再均衡后分配的分区
使用：
将消费位移保存在一个局部变量中，如果正常消费则通过异步方法提交消费位移，如发生再均衡，则通过再均衡监听器的onPartitionsRevoked方法回调执行commitSync同步提交偏移量。
HashMap<TopicPartition, OffsetAndMetadata> currentOffsets = new HashMap<>();
consumer.subscribe(Arrays.asList(topic), new ConsumerRebalanceListener()
@Override
public void onPartitionsRevoked(Collection<TopicPartition> partitions)
consumer.commitSync(currentOffsets);
currentOffsets.clear();
@Override
public void onPartitionsAssigned(Collection<TopicPartition> partitions)
);
同时，也可以结合外部数据源管理消费位移：
consumer.subscribe(Arrays.asList(topic), new ConsumerRebalanceListener()
@Override
public void onPartitionsRevoked(Collection<TopicPartition> partitions)
//store offset in DB
@Override
public void onPartitionsAssigned(Collection<TopicPartition> partitions)
for (TopicPartition partition : partitions)
consumer.seek(partition,getOffsetFromDB(partition)); //从DB中读取消费位移
);
以上是关于Kafka:消费者详解的主要内容，如果未能解决你的问题，请参考以下文章