Kafka 消费过程和原理

使用 Kafka Client 来实现消费过程

Properties properties = new Properties();
properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "127.0.0.1:9092");
properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, " org.apache.kafka.common.serialization.StringDeserializer");
properties.put(ConsumerConfig.GROUP_ID_CONFIG, "consumer-group");
Duration timeout = Duration.ofMillis(100);
try (KafkaConsumer<String, String> consumer = new KafkaConsumer<>(properties)) {
  consumer.subscribe(List.of(TOPIC, TOPIC2));
  while (true) {
    ConsumerRecords<String, String> consumerRecords = consumer.poll(timeout);
    for (ConsumerRecord<String, String> record : consumerRecords) {
      System.out.printf("topic = %s, partition = %d, offset = %d, " + "key = %s, value = %s\n",
        record.topic(), record.partition(), record.offset(), record.key(), record.value());
    }
  }
}

Kafka 消费者的并发设计

每个消费者一个线程

private static void startConsumer() {
    for (int i = 0; i < 4; i++) {
      Thread t = new Thread(() -> consume(), "Thread-consumer-" + i);
      t.start();
    }
  }

  private static void consume() {
		// 此处的 properties 也不能在线程之间共享，否则无法启动 4 个 consumer
    Properties properties = new Properties();
    properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "127.0.0.1:9092");
    properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,
        "org.apache.kafka.common.serialization.StringDeserializer");
    properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,
        " org.apache.kafka.common.serialization.StringDeserializer");
    properties.put(ConsumerConfig.GROUP_ID_CONFIG, "thread-pre-consumer");
    Duration timeout = Duration.ofMillis(100);
    try (KafkaConsumer<String, String> consumer = new KafkaConsumer<>(properties)) {
      consumer.subscribe(List.of(TOPIC));
      //不停拉取新事件
			while (true) {
        ConsumerRecords<String, String> consumerRecords = consumer.poll(timeout);
        for (ConsumerRecord<String, String> record : consumerRecords) {
          System.out.printf("Thread = %s, topic = %s, partition = %d, offset = %d, " + "customer = %s, country = %s\n",
              Thread.currentThread().getName(), record.topic(), record.partition(), record.offset(), record.key(),
              record.value());
        }
      }
    }
  }

• max.poll.records - 设置为较小的值

• max.poll.interval.ms - 设置为更高的值

每个消费者多个线程

• 拆分主线程一个循环内的任务，解耦三个阶段

// MultiThreadConsumerWrapper.java
public void consume(Properties properties) {
    Duration timeout = Duration.ofMillis(100);
    try {
      consumer = new KafkaConsumer<>((Properties) properties.clone());
      consumer.subscribe(List.of("multi-thread-topic"));
      //一个执行循环 ♻️
			while (!stopped.get()) {
        //处理拉取到的消息
        handleRecords(consumer.poll(timeout));
				//检查任务是不是并发执行完毕
        checkTasks();
				//提交任务执行中产生的偏移量
        commitOffsets();
      }
    } catch (WakeupException e) {
      if (!stopped.get()) {
        throw new IllegalStateException(e);
      }
    } finally {
      if (consumer != null) {
        consumer.close();
      }
    }
  }

• 多线程并发处理消息

private void handleRecords(ConsumerRecords<String, String> records) {
    if (records.count() > 0) {
      List<TopicPartition> needPause = new ArrayList<>();
			//此处将每个分区放入一个线程执行
      records.partitions().forEach(partition -> {
        List<ConsumerRecord<String, String>> partitionRecords = records.records(partition);
        Task task = new Task(partitionRecords, record -> System.out.printf(
            "Thread = %s, topic = %s, partition = %d, offset = %d, " + "customer = %s, country = %s\n",
            Thread.currentThread().getName(), record.topic(), record.partition(), record.offset(), record.key(),
            record.value()));
        needPause.add(partition);
        //提交到线程池中执行
        executorService.submit(task);
        //记录任务和分区对应关系
        activeTasks.put(partition, task);
      });
			//任务是异步处理的，所以为了保持同个分区的顺序性，可以将此分区暂停拉取，此时 poll 方法不会返回此分区的任何数据
      consumer.pause(needPause);
    }
  }

• 检查任务执行情况

private void checkTasks() {
    List<TopicPartition> finishedPartitions = new ArrayList<>();
    activeTasks.forEach((partition, task) -> {
      if (task.isFinished()) {
        finishedPartitions.add(partition);
      }
      long offset = task.getCurrentOffset();
      if (offset > 0) {
				//记录当前分区的偏移量，统一提交
        offsetsToCommit.put(partition, new OffsetAndMetadata(offset));
      }
    });
    finishedPartitions.forEach(activeTasks::remove);
		//处理完成之后可以继续拉取下一批数据
    consumer.resume(finishedPartitions);
  }

• 提交偏移量

private void commitOffsets() {
    try {
      long currentTimeMillis = System.currentTimeMillis();
			// 不会每次 poll 都提交，固定一个时间间隔提交，降低对吞吐量的影响
      if (currentTimeMillis - lastCommitTime > COMMIT_INTERVAL) {
        if (!offsetsToCommit.isEmpty()) {
          consumer.commitSync(offsetsToCommit);
          offsetsToCommit.clear();
        }
        lastCommitTime = currentTimeMillis;
      }
    } catch (Exception e) {
      System.out.println("Failed to commit offsets -- " + e.getMessage());
    }
  }

• 监听再均衡事件，等待正在执行的任务完成，剩余的消息不再处理

// public class MultiThreadConsumerWrapper implements ConsumerRebalanceListener
	@Override
  public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
    // 停止所有被撤销所有权的分区上的任务
    Map<TopicPartition, Task> stoppedTask = new HashMap<>();
    for (TopicPartition partition : partitions) {
      Task task = activeTasks.remove(partition);
      if (task != null) {
        task.stop();
        stoppedTask.put(partition, task);
      }
    }

    // 如果被撤销的分区，有正在进行的任务，则等待其完成，记录偏移量
    stoppedTask.forEach((partition, task) -> {
      long offset = task.waitForCompletion();
      if (offset > 0) {
        offsetsToCommit.put(partition, new OffsetAndMetadata(offset));
      }
    });

    // 提交所有可以提交的偏移量
    Map<TopicPartition, OffsetAndMetadata> offsetNeedCommit = new HashMap<>();
    partitions.forEach(partition -> {
      OffsetAndMetadata offset = offsetsToCommit.remove(partition);
      if (offset != null) {
        offsetNeedCommit.put(partition, offset);
      }
    });
    try {
      consumer.commitSync(offsetNeedCommit);
    } catch (Exception e) {
      System.out.println("Failed to commit offsets for revoked partitions!");
    }
  }

• 停止消费者

public void stop() {
    stopped.set(true);
		//这个方法是线程安全的，可以唤醒正在阻塞的 poll 方法，抛出 WakeupException
    consumer.wakeup();
}

Spring 中如何实现消费者的并发消费

@KafkaListener(id = "consumer", topics = TOPIC, groupId = "spring-1", autoStartup = "false", concurrency = "4")
  public void listen1(String value,
      @Header(KafkaHeaders.RECEIVED_TOPIC) String topic,
      @Header(KafkaHeaders.RECEIVED_KEY) String key) {
    logger.info("Consumer consumed event -- thread:{}, topic: {}, key:{}, value:{}", Thread.currentThread().getName(),
        topic, key, value);
  }