Spring Boot Kafka - 用Multi-Consumer应对高吞吐量

ゝ一纸荒年。 2022-09-05 12:50 5阅读 0赞

### 文章目录 ###

*  Spring Boot Kafka - 用Multi-Consumer应对高吞吐量
 *   *  前言
     *  主题与分区
     *  低吞吐量场景
     *  高吞吐量场景
     *  增加Topic的分区数
     *  使用Multi Consumer
     *  参考文档

# Spring Boot Kafka - 用Multi-Consumer应对高吞吐量 #

## 前言 ##

前面几篇文章讲到了Spring Kafka如何来发送消息和接收消息。

本文描述了如何用Multi-Consumer应对高吞吐量的场景。

## 主题与分区 ##

Kafka的消息通过Topic（主题）进行分类。主题就好比文件系统的文件夹。主题可以被分为若干个分区（Partition），一个分区就是一个提交日志。消息以追加的方式写入分区，然后以先入先出的顺序读取。

由于一个主题一般包含几个分区，因此无法在整个主题范围内保证消息的顺序，但Kafka可以保证消息在单个分区内的顺序。

生产者在默认情况下把消息均衡地分布到主题的所有分区上，而并不关心特定消息会被写到哪个分区。不过，Kafka保证有相同message key的消息会被写入到同一个分区。

## 低吞吐量场景 ##

在默认情况，一个Topic只有一个Partition。Producer(s)将消息添加到该Partition的末尾，一个Consumer从该Partition中依序读走消息去处理。

比喻：

*  在访问量不大的网站，只要一台服务器就可以正常提供服务。

## 高吞吐量场景 ##

在高吞吐量时：

*  为了加快写入消息的速度，需要将一个Topic分成多个分区
 *  为了加快读取消息的速度，需要有多个Consumer同时来从分区中读取消息
 *  用Consumer Group来保证一个分区只能被一个Consumer读取 （也就是说在一个Consumer Group内一个消息只能被读取一次)

注意：

*  在一个Consumer Group内，一个Consumer可以读取多个分区，但是一个分区只能被一个Consumer读取。

比喻：

*  在访问量很大的网站，通常需要负载均衡和多台服务器才能正常提供服务。

## 增加Topic的分区数 ##

默认情况下Topic的分区数为1，为了应对高吞吐量的场景，可以增加Topic的分区数。

因为Topic的分区是有状态的，因此增加分区后不能减少，因此增加分区时需要慎重。

# 查看`users`主题的详情，包括分区数等
    docker exec -it kafka_kafka_1 \
      kafka-topics.sh --bootstrap-server localhost:9092 --describe --topic users
      
    # 将`users`主题改为10个分区
    docker exec -it kafka_kafka_1 \
      kafka-topics.sh --bootstrap-server localhost:9092 --alter --topic users  --partitions 10

## 使用Multi Consumer ##

在Spring Boot应用配置`application.yaml`中配置Multi Consumer。

server:
      port: 8080
    spring:
      kafka:
        listener:
          concurrency: 3
        consumer:
          bootstrap-servers: localhost:9092
          group-id: demo-group
          auto-offset-reset: earliest
          key-deserializer: org.springframework.kafka.support.serializer.ErrorHandlingDeserializer
          value-deserializer: org.springframework.kafka.support.serializer.ErrorHandlingDeserializer
          properties:
            spring.json.trusted.packages: '*'
            spring.deserializer.key.delegate.class: org.apache.kafka.common.serialization.StringDeserializer
            spring.deserializer.value.delegate.class: org.springframework.kafka.support.serializer.JsonDeserializer
            partition.assignment.strategy: org.apache.kafka.clients.consumer.RoundRobinAssignor
        producer:
          bootstrap-servers: localhost:9092
          key-serializer: org.apache.kafka.common.serialization.StringSerializer
          value-serializer: org.springframework.kafka.support.serializer.JsonSerializer

说明：

*  `consumer.group-id: demo-group` \- Consumer Group ID为`demo-group`
 *  `listener.concurrency: 3` \- 配置Consumer数为3个
 *  `partition.assignment.strategy` 配置为`org.apache.kafka.clients.consumer.RoundRobinAssignor` ，使得在一个Consumer侦听多个主题时，每个Consumer都被分配到某个分区，避免Consumer处在空闲状态。默认值为`PartitionAssignor` ，只适合一个Consumer侦听一个主题的情况。

重新启动Spring Boot应用后，查看Consumer Group详情：

# 查看Consumer Group详情
    docker exec -it kafka_kafka_1 \
      kafka-consumer-groups.sh --bootstrap-server kafka:9092 --describe  --group demo-group

可以看到一共有3个Consumer来连接10个Partition，其中1个Consumer连接4个Partition，另外2个Consumer分别连接3个Partition。

注意，在配置分区和Consumer数量时，一定要设置分区数 >= Consumer数，让一个Consumer至少读取一个分区，不能存在空闲的Consumer，否则就达不到提高吞吐量的目的。

## 参考文档 ##

*  [Spring Boot and Kafka Configuration Tuning][]
 *  [Spring Kafka - Using ConcurrentMessageListenerContainer][]
 *  [Kafka Introduction][]

[Spring Boot and Kafka Configuration Tuning]: https://dzone.com/articles/spring-boot-and-kafka-configuration-tuning
[Spring Kafka - Using ConcurrentMessageListenerContainer]: https://docs.spring.io/spring-kafka/reference/html/#using-ConcurrentMessageListenerContainer
[Kafka Introduction]: https://kafka.apache.org/intro