kafka消费者客户端

Kafka消费者

1.1 消费者与消费者组

消费者与消费者组之间的关系

​ 每一个消费者都隶属于某一个消费者组，一个消费者组可以包含一个或多个消费者，每一条消息只会被消费者组中的某一个消费者所消费。不同消费者组之间消息的消费是互不干扰的。

为什么会有消费者组的概念

​ 消费者组出现主要是出于两个目的:

​ (1) 使整体的消费能力具备横向的伸缩性。可以适当增加消费者组中消费者的数量，来提高整体的消费能力。但是每一个分区至多被消费者组的中一个消费者所消费，因此当消费者组中消费者数量超过分区数时，多出的消费者不会分配到任何一个分区。当然这是默认的分区分配策略，可通过partition.assignment.strategy进行配置。

​ (2) 实现消息消费的隔离。不同消费者组之间消息消费互不干扰，从而实现发布订阅这种消息投递模式。

注意：

​ 消费者隶属的消费者组可以通过group.id进行配置。消费者组是一个逻辑上的概念，但消费者并不是一个逻辑上的概念，它可以是一个线程，也可以是一个进程。同一个消费者组内的消费者可以部署在同一台机器上，也可以部署在不同的机器上。

1.2 消费者客户端开发

​ 一个正常的消费逻辑需要具备以下几个步骤：

• 配置消费者客户端参数及创建相应的消费者实例。

• 订阅主题

• 拉取消息并消费

• 提交消费位移

• 关闭消费者实例

  public class KafkaConsumerAnalysis {
      public static final String brokerList="node112:9092,node113:9092,node114:9092";
      public static final String topic = "topic-demo";
      public static final String groupId = "group.demo";
      public static final AtomicBoolean isRunning = new AtomicBoolean(true);
  ​
      public static Properties initConfig() {
          Properties prop = new Properties();
          prop.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, brokerList);
          prop.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
          prop.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
          prop.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);
          prop.put(ConsumerConfig.CLIENT_DNS_LOOKUP_CONFIG, "consumer.client.di.demo");
          return prop;
      }
  ​
  ​
      public static void main(String[] args) {
          KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(initConfig());
          
          for (ConsumerRecord<String, String> record : records) {
              System.out.println("topic = " + record.topic() + ", partition =" +                                               record.partition() + ", offset = " + record.offset());
              System.out.println("key = " + record.key() + ", value = " + record.value());
                  }
              }
          } catch (Exception e) {
              e.printStackTrace();
          }finally {
              consumer.close();
          }
      }
  }
  ​

1.2.1 订阅主题和分区

​ 先来说一下消费者订阅消息的粒度：一个消费者可以订阅一个主题、多个主题、或者多个主题的特定分区。主要通过subsribe和assign两个方法实现订阅。

（1）订阅一个主题：

​ public void subscribe(Collection<String> topics)，当集合中有一个主题时。

（2）订阅多个主题：

​ public void subscribe(Collection<String> topics)，当集合中有多个主题时。

​ public void subscribe(Pattern pattern)，通过正则表达式实现消费者主题的匹配。通过这种方式，如果在消息消费的过程中，又添加了新的能够匹配到正则的主题，那么消费者就可以消费到新添加的主题。 consumer.subscribe(Pattern.compile("topic-.*"));

（3）多个主题的特定分区

​ public void assign(Collection<TopicPartition> partitions)，可以实现订阅某些特定的主题分区。TopicPartition包括两个属性：topic（String）和partition（int）。

​ 如果事先不知道有多少分区该如何处理，KafkaConsumer中的partitionFor方法可以获得指定主题分区的元数据信息：

​ public List<PartitionInfo> partitionsFor(String topic)

​ PartitionInfo的属性如下：

  
  public class PartitionInfo {
      private final String topic;//主题
      private final int partition;//分区
      private final Node leader;//分区leader
      private final Node[] replicas;//分区的AR
      private final Node[] inSyncReplicas;//分区的ISR
      private final Node[] offlineReplicas;//分区的OSR
   }

​ 因此也可以通过这个方法实现某个主题的全部订阅。

​ 需要指出的是，subscribe(Collection)、subscirbe(Pattern)、assign(Collection)方法分别代表了三种不同的订阅状态：AUTO_TOPICS、AUTO_PATTREN和USER_ASSIGN，这三种方式是互斥的，消费者只能使用其中一种，否则会报出IllegalStateException。

​ subscirbe方法可以实现消费者自动再平衡的功能。多个消费者的情况下，可以根据分区分配策略自动分配消费者和分区的关系，当消费者增加或减少时，也能实现负载均衡和故障转移。

​ 如何实现取消订阅：

​ consumer.unsubscribe()

1.2.2 反序列化

​ KafkaProducer端生产消息进行序列化，同样消费者就要进行相应的反序列化。相当于根据定义的序列化格式的一个逆序提取数据的过程。

  
  import com.gdy.kafka.producer.Company;
  import org.apache.kafka.common.errors.SerializationException;
  import org.apache.kafka.common.serialization.Deserializer;
  ​
  import java.io.UnsupportedEncodingException;
  import java.nio.ByteBuffer;
  import java.util.Map;
  ​
  public class CompanyDeserializer implements Deserializer<Company> {
      @Override
      public void configure(Map<String, ?> configs, boolean isKey) {
  ​
      }
  ​
      @Override
      public Company deserialize(String topic, byte[] data) {
          if(data == null) {
              return null;
          }
  ​
          if(data.length < 8) {
              throw new SerializationException("size of data received by Deserializer is shorter than expected");
          }
  ​
          ByteBuffer buffer = ByteBuffer.wrap(data);
          int nameLength = buffer.getInt();
          byte[] nameBytes = new byte[nameLength];
          buffer.get(nameBytes);
          int addressLen = buffer.getInt();
          byte[] addressBytes = new byte[addressLen];
          buffer.get(addressBytes);
          String name,address;
          try {
              name = new String(nameBytes,"UTF-8");
              address = new String(addressBytes,"UTF-8");
          }catch (UnsupportedEncodingException e) {
              throw new SerializationException("Error accur when deserializing");
          }
  ​
          return new Company(name, address);
      }
  ​
      @Override
      public void close() {
  ​
      }
  }

​ 实际生产中需要自定义序列化器和反序列化器时，推荐使用Avro、JSON、Thrift、ProtoBuf或者Protostuff等通用的序列化工具来包装。

1.2.3 消息消费

​ Kafka中消息的消费是基于拉模式的，kafka消息的消费是一个不断轮旋的过程，消费者需要做的就是重复的调用poll方法。

  
  public ConsumerRecords<K, V> poll(final Duration timeout)

​ 这个方法需要注意的是，如果消费者的缓冲区中有可用的数据，则会立即返回，否则会阻塞至timeout。如果在阻塞时间内缓冲区仍没有数据，则返回一个空的消息集。timeout的设置取决于应用程序对效应速度的要求。如果应用线程的位移工作是从Kafka中拉取数据并进行消费可以将这个参数设置为Long.MAX_VALUE。

​ 每次poll都会返回一个ConsumerRecords对象，它是ConsumerRecord的集合。对于ConsumerRecord相比于ProducerRecord多了一些属性：

  
   private final String topic;//主题
      private final int partition;//分区
      private final long offset;//偏移量
      private final long timestamp;//时间戳
      private final TimestampType timestampType;//时间戳类型
      private final int serializedKeySize;//序列化key的大小
      private final int serializedValueSize;//序列化value的大小
      private final Headers headers;//headers
      private final K key;//key
      private final V value;//value
      private volatile Long checksum;//CRC32校验和

​ 另外我们可以按照分区维度对消息进行消费，通过ConsumerRecords.records(TopicPartiton)方法实现。

  
   ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000));
       Set<TopicPartition> partitions = records.partitions();
       for (TopicPartition tp : partitions) {
            for (ConsumerRecord<String, String> record : records.records(tp)) {
                 System.out.println(record.partition() + " ," + record.value());
            }
        }

​ 另外还可以按照主题维度对消息进行消费，通过ConsumerRecords.records(Topic)实现。

  
  for (String topic : topicList) {
         for (ConsumerRecord<String, String> record : records.records(topic)) {
                  System.out.println(record.partition() + " ," + record.value());
          }
   }

1.2.4 消费者位移提交

​ 首先要 明白一点，消费者位移是要做持久化处理的，否则当发生消费者崩溃或者消费者重平衡时，消费者消费位移无法获得。旧消费者客户端是将位移提交到zookeeper上，新消费者客户端将位移存储在Kafka内部主题_consumer_offsets中。

​ KafkaConsumer提供了两个方法position(TopicPatition)和commited(TopicPartition)。

​ public long position(TopicPartition partition)-----获得下一次拉取数据的偏移量

​ public OffsetAndMetadata committed(TopicPartition partition)-----给定分区的最后一次提交的偏移量。

还有一个概念称之为lastConsumedOffset，这个指的是最后一次消费的偏移量。

​ 在kafka提交方式有两种：自动提交和手动提交。

（1）自动位移提交

​ kafka默认情况下采用自动提交，enable.auto.commit的默认值为true。当然自动提交并不是没消费一次消息就进行提交，而是定期提交，这个定期的周期时间由auto.commit.intervals.ms参数进行配置，默认值为5s，当然这个参数生效的前提就是开启自动提交。

​ 自动提交会造成重复消费和消息丢失的情况。重复消费很容易理解，因为自动提交实际是延迟提交，因此很容易造成重复消费，然后消息丢失是怎么产生的？

（2）手动位移提交

​ 开始手动提交的需要配置enable.auto.commit=false。手动提交消费者偏移量，又可分为同步提交和异步提交。

​ 同步提交：

​ 同步提交很简单，调用commitSync() 方法：

  
   while (isRunning.get()) {
           ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000));
           for (ConsumerRecord<String, String> record : records) {
               //consume message
               consumer.commitSync();
            }
   }

​ 这样，每消费一条消息，提交一个偏移量。当然可用过缓存消息的方式，实现批量处理+批量提交：

  
  while (isRunning.get()) {
          ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000));
          for (ConsumerRecord<String, String> record : records) {
               buffer.add(record);
          }
          if (buffer.size() >= minBaches) {
              for (ConsumerRecord<String, String> record : records) {
                  //consume message
              }
              consumer.commitSync();
              buffer.clear();
          }
    }

​ 还可以通过public void commitSync(final Map<TopicPartition, OffsetAndMetadata> offsets)这个方法实现按照分区粒度进行同步提交。

  
  while (isRunning.get()) {
      ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000));
      for (TopicPartition tp : records.partitions()) {
          List<ConsumerRecord<String, String>> partitionRecords = records.records(tp);
          for (ConsumerRecord record : partitionRecords) {
              //consume message
          }
          long lastConsumerOffset = partitionRecords.get(partitionRecords.size() - 1).offset();
          consumer.commitSync(Collections.singletonMap(tp，new                                                               OffsetAndMetadata(lastConsumerOffset+1)));
      }
  }

​ 异步提交：

​ commitAsync异步提交的时候消费者线程不会被阻塞，即可能在提交偏移量的结果还未返回之前，就开始了新一次的拉取数据操作。异步提交可以提升消费者的性能。commitAsync有三个重载：

​ public void commitAsync()

​ public void commitAsync(OffsetCommitCallback callback)

​ public void commitAsync(final Map<TopicPartition, OffsetAndMetadata> offsets, OffsetCommitCallback )

​ 对照同步提交的方法参数，多了一个Callback回调参数，它提供了一个异步提交的回调方法，当消费者位移提交完成后回调OffsetCommitCallback的onComplement方法。以第二个方法为例：

  
  ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000));
      for (ConsumerRecord<String, String> record : records) {
          //consume message
      }
      consumer.commitAsync(new OffsetCommitCallback() {
          @Override
          public void onComplete(Map<TopicPartition, OffsetAndMetadata> offsets, Exception e) {
             if (e == null) {
                  System.out.println(offsets);
              }else {
                   e.printStackTrace();
              }
          }
  });

1.2.5 控制和关闭消费

​ kafkaConsumer提供了pause()和resume() 方法分别实现暂停某些分区在拉取操作时返回数据给客户端和恢复某些分区向客户端返回数据的操作：

​ public void pause(Collection<TopicPartition> partitions)

​ public void resume(Collection<TopicPartition> partitions)

​ 优雅停止KafkaConsumer退出消费者循环的方式：

​ （1）不要使用while(true)，而是使用while(isRunning.get())，isRunning是一个AtomicBoolean类型，可以在其他地方调用isRunning.set(false)方法退出循环。

​ （2）调用consumer.wakup()方法，wakeup方法是KafkaConsumer中唯一一个可以从其他线程里安全调用的方法，会抛出WakeupException，我们不需要处理这个异常。

​ 跳出循环后一定要显示的执行关闭动作和释放资源。

1.2.6 指定位移消费

KafkaConsumer可通过两种方式实现实现不同粒度的指定位移消费。第一种是通过auto.offset.reset参数，另一种通过一个重要的方法seek。

（1）auto.offset.reset

auto.offset.reset这个参数总共有三种可配置的值：latest、earliest、none。如果配置不在这三个值当中，就会抛出ConfigException。

latest：当各分区下有已提交的offset时，从提交的offset开始消费；无提交的offset或位移越界时，消费新产生的该分区下的数据

earliest：当各分区下有已提交的offset时，从提交的offset开始消费；无提交的offset或位移越界时，从头开始消费

none：topic各分区都存在已提交的offset时，从offset后开始消费；只要有一个分区不存在已提交的offset或位移越界，则抛出NoOffsetForPartitionException异常

消息的消费是通过poll方法进行的，poll方法对于开发者来说就是一个黑盒，无法精确的掌控消费的起始位置。即使通过auto.offsets.reset参数也只能在找不到位移或者位移越界的情况下粗粒度的从头开始或者从末尾开始。因此，Kafka提供了另一种更细粒度的消费掌控：seek。

（2）seek

seek可以实现追前消费和回溯消费：

  
   public void seek(TopicPartition partition, long offset) 

可以通过seek方法实现指定分区的消费位移的控制。需要注意的一点是，seek方法只能重置消费者分配到的分区的偏移量，而分区的分配是在poll方法中实现的。因此在执行seek方法之前需要先执行一次poll方法获取消费者分配到的分区，但是并不是每次poll方法都能获得数据，所以可以采用如下的方法。

  
   consumer.subscribe(topicList);
      Set<TopicPartition> assignment = new HashSet<>();
      while(assignment.size() == 0) {
          consumer.poll(Duration.ofMillis(100));
          assignment = consumer.assignment();//获取消费者分配到的分区，没有获取返回一个空集合
      }
  ​
      for (TopicPartition tp : assignment) {
          consumer.seek(tp, 10); //重置指定分区的位移
      }
      while (true) {
          ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000));
          //consume record
       }

如果对未分配到的分区执行了seek方法，那么会报出IllegalStateException异常。

在前面我们已经提到，使用auto.offsets.reset参数时，只有当消费者分配到的分区没有提交的位移或者位移越界时，才能从earliest消费或者从latest消费。seek方法可以弥补这一中情况，实现任意情况的从头或从尾部消费。

   Set<TopicPartition> assignment = new HashSet<>();
      while(assignment.size() == 0) {
          consumer.poll(Duration.ofMillis(100));
          assignment = consumer.assignment();
      }
      Map<TopicPartition, Long> offsets = consumer.endOffsets(assignment);//获取指定分区的末尾位置
      for (TopicPartition tp : assignment) {
          consumer.seek;
      }

与endOffset对应的方法是beginningOffset方法，可以获取指定分区的起始位置。其实kafka已经提供了一个从头和从尾消费的方法。

  
  public void seekToBeginning(Collection<TopicPartition> partitions)
  public void seekToEnd(Collection<TopicPartition> partitions)

还有一种场景是这样的，我们并不知道特定的消费位置，却知道一个相关的时间点。为解决这种场景遇到的问题，kafka提供了一个offsetsForTimes()方法，通过时间戳来查询分区消费的位移。

      Map<TopicPartition, Long> timestampToSearch = new HashMap<>();
      for (TopicPartition tp : assignment) {
          timestampToSearch.put(tp, System.currentTimeMillis() - 24 * 3600 * 1000);
      }
   //获得指定分区指定时间点的消费位移
      Map<TopicPartition, OffsetAndTimestamp> offsets =                                                                                   consumer.offsetsForTimes(timestampToSearch);
      for (TopicPartition tp : assignment) {
          OffsetAndTimestamp offsetAndTimestamp = offsets.get(tp);
          if (offsetAndTimestamp != null) {
                  consumer.seek(tp, offsetAndTimestamp.offset());
          }
      }

由于seek方法的存在，使得消费者的消费位移可以存储在任意的存储介质中，包括DB、文件系统等。

1.2.7 消费者的再均衡

再均衡是指分区的所属权从一个消费者转移到另一消费者的行为，它为消费者组具备高可用和伸缩性提高保障。不过需要注意的地方有两点，第一是消费者发生再均衡期间，消费者组中的消费者是无法读取消息的。第二点就是消费者发生再均衡可能会引起重复消费问题，所以一般情况下要尽量避免不必要的再均衡。

KafkaConsumer的subscribe方法中有一个参数为ConsumerRebalanceListener，我们称之为再均衡监听器，它可以用来在设置发生再均衡动作前后的一些准备和收尾动作。

  public interface ConsumerRebalanceListener {
      void onPartitionsRevoked(Collection<TopicPartition> partitions);
      void onPartitionsAssigned(Collection<TopicPartition> partitions);
  }

onPartitionsRevoked方法会在再均衡之前和消费者停止读取消息之后被调用。可以通过这个回调函数来处理消费位移的提交，以避免重复消费。参数partitions表示再均衡前分配到的分区。

onPartitionsAssigned方法会在再均衡之后和消费者消费之间进行调用。参数partitons表示再均衡之后所分配到的分区。

  consumer.subscribe(topicList);
      Map<TopicPartition, OffsetAndMetadata> currentOffsets = new HashMap<>();
      consumer.subscribe(topicList, new ConsumerRebalanceListener() {
          @Override
          public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
              consumer.commitSync(currentOffsets);//提交偏移量
          }
  ​
          @Override
          public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
              //do something
          }
      });
  ​
      try {
          while (isRunning.get()) {
              ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000));
              for (ConsumerRecord<String, String> record : records) {
                  //process records
                  //记录当前的偏移量
                  currentOffsets.put(new TopicPartition(record.topic(), record.partition()),new                                OffsetAndMetadata( record.offset() + 1));
              }
              consumer.commitAsync(currentOffsets, null);
          }
  ​
          } catch (Exception e) {
              e.printStackTrace();
          }finally {
              consumer.close();
          }

1.2.8 消费者拦截器

消费者拦截器主要是在消费到消息或者提交消费位移时进行一些定制化的操作。消费者拦截器需要自定义实现org.apache.kafka.clients.consumer.ConsumerInterceptor接口。

  public interface ConsumerInterceptor<K, V> extends Configurable {    
      public ConsumerRecords<K, V> onConsume(ConsumerRecords<K, V> records);
      public void onCommit(Map<TopicPartition, OffsetAndMetadata> offsets);
      public void close();
  }

onConsume方法是在poll()方法返回之前被调用，比如修改消息的内容、过滤消息等。如果onConsume方法发生异常，异常会被捕获并记录到日志中，但是不会向上传递。

Kafka会在提交位移之后调用拦截器的onCommit方法，可以使用这个方法来记录和跟踪消费的位移信息。

  
  public class ConsumerInterceptorTTL implements ConsumerInterceptor<String,String> {
      private static final long EXPIRE_INTERVAL = 10 * 1000; //10秒过期
      @Override
      public ConsumerRecords<String, String> onConsume(ConsumerRecords<String, String> records) {
          long now = System.currentTimeMillis();
          Map<TopicPartition, List<ConsumerRecord<String, String>>> newRecords = new HashMap<>();
  ​
          for (TopicPartition tp : records.partitions()) {
              List<ConsumerRecord<String, String>> tpRecords = records.records(tp);
              List<ConsumerRecord<String, String>> newTpRecords = records.records(tp);
              for (ConsumerRecord<String, String> record : tpRecords) {
                  if (now - record.timestamp() < EXPIRE_INTERVAL) {//判断是否超时
                      newTpRecords.add(record);
                  }
              }
              if (!newRecords.isEmpty()) {
                  newRecords.put(tp, newTpRecords);
              }
  ​
  ​
          }
          return new ConsumerRecords<>(newRecords);
      }
  ​
      @Override
      public void onCommit(Map<TopicPartition, OffsetAndMetadata> offsets) {
          offsets.forEach((tp,offset) -> {
              System.out.println(tp + ":" + offset.offset());
          });
      }
  ​
      @Override
      public void close() {}
  ​
      @Override
      public void configure(Map<String, ?> configs) {}
  }

使用这种TTL需要注意的是如果采用带参数的位移提交方式，有可能提交了错误的位移，可能poll拉取的最大位移已经被拦截器过滤掉。

1.2.9 消费者的多线程实现

KafkaProducer是线程安全的，然而KafkaConsumer是非线程安全的。KafkaConsumer中的acquire方法用于检测当前是否只有一个线程在操作，如果有就会抛出ConcurrentModifiedException。acuqire方法和我们通常所说的锁是不同的，它不会阻塞线程，我们可以把它看做是一个轻量级的锁，它通过线程操作计数标记的方式来检测是否发生了并发操作。acquire方法和release方法成对出现，分表表示加锁和解锁。

  //标记当前正在操作consumer的线程
  private final AtomicLong currentThread = new AtomicLong(NO_CURRENT_THREAD);
  //refcount is used to allow reentrant access by the thread who has acquired currentThread,
  //大概可以理解我加锁的次数
  private final AtomicInteger refcount = new AtomicInteger(0);
  private void acquire() {
   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();
  }
  ​
  private void release() {
      if (refcount.decrementAndGet() == 0)
          currentThread.set(NO_CURRENT_THREAD);
  }

kafkaConsumer中的每个共有方法在调用之前都会执行aquire方法，只有wakeup方法是个意外。

KafkaConsumer的非线程安全并不意味着消费消息的时候只能以单线程的方式执行。可以通过多种方式实现多线程消费。

（1）Kafka多线程消费第一种实现方式--------线程封锁

所谓线程封锁，就是为每个线程实例化一个KafkaConsumer对象。这种方式一个线程对应一个KafkaConsumer，一个线程（可就是一个consumer)可以消费一个或多个分区的消息。这种消费方式的并发度受限于分区的实际数量。当线程数量超过分分区数量时，就会出现线程限制额的情况。

  import org.apache.kafka.clients.consumer.ConsumerConfig;
  import org.apache.kafka.clients.consumer.ConsumerRecord;
  import org.apache.kafka.clients.consumer.ConsumerRecords;
  import org.apache.kafka.clients.consumer.KafkaConsumer;
  import org.apache.kafka.common.serialization.StringDeserializer;
  import java.time.Duration;
  import java.util.Arrays;
  import java.util.Properties;
  ​
  public class FirstMutiConsumerDemo  {
      public static final String brokerList="node112:9092,node113:9092,node114:9092";
      public static final String topic = "topic-demo";
      public static final String groupId = "group.demo";
  ​
      public static Properties initConfig() {
          Properties prop = new Properties();
          prop.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, brokerList);
          prop.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,                  StringDeserializer.class.getName());
          prop.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,                 StringDeserializer.class.getName());
          prop.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);
          prop.put(ConsumerConfig.CLIENT_DNS_LOOKUP_CONFIG, "consumer.client.di.demo");
          prop.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, true);
          return prop;
      }
  ​
      public static void main(String[] args) {
          Properties prop = initConfig();
          int consumerThreadNum = 4;
          for (int i = 0; i < 4; i++) {
              new KafkaCoosumerThread(prop, topic).run();
          }
      }
  ​
      public static class KafkaCoosumerThread extends Thread {
       //每个消费者线程包含一个KakfaConsumer对象。
          private KafkaConsumer<String, String> kafkaConsumer;
          public KafkaCoosumerThread(Properties prop, String topic) {
              this.kafkaConsumer = new KafkaConsumer<String, String>(prop);
              this.kafkaConsumer.subscribe(Arrays.asList(topic));
          }
  ​
          @Override
          public void run() {
              try {
                  while (true) {
                      ConsumerRecords<String, String> records =                   kafkaConsumer.poll(Duration.ofMillis(100));
                      for (ConsumerRecord<String, String> record : records) {
                          //处理消息模块
                      }
                  }
              } catch (Exception e) {
                  e.printStackTrace();
              }finally {
                  kafkaConsumer.close();
              }
          }
      }
  }
  ​

这种实现方式和开启多个消费进程的方式没有本质的区别，优点是每个线程可以按照顺序消费消费各个分区的消息。缺点是每个消费线程都要维护一个独立的TCP连接，如果分区数和线程数都很多，那么会造成不小的系统开销。

（2）Kafka多线程消费第二种实现方式--------多个消费线程同时消费同一分区

多个线程同时消费同一分区，通过assign方法和seek方法实现。这样就可以打破原有消费线程个数不能超过分区数的限制，进一步提高了消费的能力，但是这种方式对于位移提交和顺序控制的处理就会变得非常复杂。实际生产中很少使用。

（3）第三种实现方式-------创建一个消费者，records的处理使用多线程实现

一般而言，消费者通过poll拉取数据的速度相当快，而整体消费能力的瓶颈也正式在消息处理这一块。基于此

考虑第三种实现方式。

  import org.apache.kafka.clients.consumer.ConsumerConfig;
  import org.apache.kafka.clients.consumer.ConsumerRecord;
  import org.apache.kafka.clients.consumer.ConsumerRecords;
  import org.apache.kafka.clients.consumer.KafkaConsumer;
  import org.apache.kafka.common.serialization.StringDeserializer;
  import java.time.Duration;
  import java.util.Arrays;
  import java.util.Properties;
  import java.util.concurrent.ArrayBlockingQueue;
  import java.util.concurrent.ExecutorService;
  import java.util.concurrent.ThreadPoolExecutor;
  import java.util.concurrent.TimeUnit;
  ​
  public class ThirdMutiConsumerThreadDemo {
      public static final String brokerList="node112:9092,node113:9092,node114:9092";
      public static final String topic = "topic-demo";
      public static final String groupId = "group.demo";
  ​
      public static Properties initConfig() {
          Properties prop = new Properties();
          prop.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, brokerList);
          prop.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,                    StringDeserializer.class.getName());
          prop.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,                   StringDeserializer.class.getName());
          prop.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);
          prop.put(ConsumerConfig.CLIENT_DNS_LOOKUP_CONFIG, "consumer.client.di.demo");
          prop.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, true);
          return prop;
      }
  ​
      public static void main(String[] args) {
          Properties prop = initConfig();
          KafkaConsumerThread consumerThread = new KafkaConsumerThread(prop, topic,              Runtime.getRuntime().availableProcessors());
          consumerThread.start();
      }
  ​
  ​
      public static class KafkaConsumerThread extends Thread {
          private KafkaConsumer<String, String> kafkaConsumer;
          private ExecutorService executorService;
          private int threadNum;
  ​
          public KafkaConsumerThread(Properties prop, String topic, int threadNum) {
              this.kafkaConsumer = new KafkaConsumer<String, String>(prop);
              kafkaConsumer.subscribe(Arrays.asList(topic));
              this.threadNum = threadNum;
              executorService = new ThreadPoolExecutor(threadNum, threadNum, 0L,                     TimeUnit.MILLISECONDS,
                new ArrayBlockingQueue<>(1000), new ThreadPoolExecutor.CallerRunsPolicy());
          }
  ​
          @Override
          public void run() {
              try {
                  while (true) {
                      ConsumerRecords<String, String> records =                   kafkaConsumer.poll(Duration.ofMillis(100));
                      if (!records.isEmpty()) {
                          executorService.submit(new RecordHandler(records));
                      }
                  }
              } catch (Exception e) {
                  e.printStackTrace();
              }finally {
                  kafkaConsumer.close();
              }
          }
      }
  ​
      public static class RecordHandler implements Runnable {
          public final ConsumerRecords<String,String> records;
          public RecordHandler(ConsumerRecords<String, String> records) {
              this.records = records;
          }
          
          @Override
          public void run() {
              //处理records
          }
      }
  }
  ​
  ​

KafkaConsumerThread类对应一个消费者线程，里面通过线程池的方式调用RecordHandler处理一批批的消息。其中线程池采用的拒绝策略为CallerRunsPolicy，当阻塞队列填满时，由调用线程处理该任务，以防止总体的消费能力跟不上poll拉取的速度。这种方式还可以进行横向扩展，通过创建多个KafkaConsumerThread实例来进一步提升整体的消费能力。

这种方式还可以减少TCP连接的数量，但是对于消息的顺序处理就变得困难了。这种方式需要引入一个共享变量Map<TopicPartition,OffsetAndMetadata> offsets参与消费者的偏移量提交。每一个RecordHandler类在处理完消息后都将对应的消费位移保存到共享变量offsets中，KafkaConsumerThread在每一次poll()方法之后都要进读取offsets中的内容并对其进行提交。对于offsets的读写要采用加锁处理，防止出现并发问题。并且在写入offsets的时候需要注意位移覆盖的问题。针对这个问题，可以将RecordHandler的run方法做如下改变：

  public void run() {
              for (TopicPartition tp : records.partitions()) {
                  List<ConsumerRecord<String, String>> tpRecords = this.records.records(tp);
                  //处理tpRecords
                  long lastConsumedOffset = tpRecords.get(tpRecords.size() - 1).offset();
                  synchronized (offsets) {
                      if (offsets.containsKey(tp)) {
                          offsets.put(tp, new OffsetAndMetadata(lastConsumedOffset + 1));
                      }else {
                          long positioin = offsets.get(tp).offset();
                          if(positioin < lastConsumedOffset + 1) {
                          offsets.put(tp, new OffsetAndMetadata(lastConsumedOffset + 1));
                          }
                      }
                  }
              }
          }

对应的位移提交代码也应该在KafkaConsumerThread的run方法中进行体现

  public void run() {
     try {
          while (true) {
              ConsumerRecords<String, String> records =                    kafkaConsumer.poll(Duration.ofMillis(100));
              if (!records.isEmpty()) {
                  executorService.submit(new RecordHandler(records));
                  synchronized (offsets) {
                      if (!offsets.isEmpty()) {
                         kafkaConsumer.commitSync(offsets);
                          offsets.clear();
                       }
                  }
              }
          }
      } catch (Exception e) {
         e.printStackTrace();
       }finally {
           kafkaConsumer.close();
         }
       }
   }

其实这种方式并不完美，可能造成数据丢失。可以通过更为复杂的滑动窗口的方式进行改进。

1.2.10 消费者重要参数

• fetch.min.bytes

  kafkaConsumer一次拉拉取请求的最小数据量。适当增加，会提高吞吐量，但会造成额外延迟。

• fetch.max.bytes

  kafkaConsumer一次拉拉取请求的最大数据量，如果kafka一条消息的大小超过这个值，仍然是可以拉取的。

• fetch.max.wait.ms

  一次拉取的最长等待时间，配合fetch.min.bytes使用

• max.partiton.fetch.bytes

  每个分区里返回consumer的最大数据量。

• max.poll.records

  一次拉取的最大消息数

• connection.max.idle.ms

  多久之后关闭限制的连接

• exclude.internal.topics

  这个参数用于设置kafka中的两个内部主题能否被公开：consumer_offsets和transaction_state。如果设为true，可以使用Pattren订阅内部主题，如果是false，则没有这种限制。

• receive.buffer.bytes

  socket接收缓冲区的大小

• send.buffer.bytes

  socket发送缓冲区的大小

• request.timeout.ms

  consumer等待请求响应的最长时间。

• reconnect.backoff.ms

  重试连接指定主机的等待时间。

• max.poll.interval.ms

  配置消费者等待拉取时间的最大值，如果超过这个期限，消费者组将剔除该消费者，进行再平衡。

• auto.offset.reset

  自动偏移量重置

• enable.auto.commit

  是否允许偏移量的自动提交

• auto.commit.interval.ms

  自动偏移量提交的时间间隔