Java Kafka分区发送及消费实战

前言

Kafka是现在非常热门的分布式消息队列,常用于微服务间异步通信，业务解耦等场景。kafka的性能非常强大，但是单个微服务吞吐性能是有上限的，我们就会用到分布式微服务，多消费者多生产者进行数据处理，保证性能同时也能根据业务量进行横向拓展，对于同一个微服务的多个实例，输入输出的topic是同一个，这时候我们就可以利用Kafka分区消费来解决这个问题。

业务场景

我们开发的是一个物联网系统，大量设备接入到平台实时发送数据，有秒级数据和分钟级别数据等等，处理流程包含接入、处理、存储，这三个模块间就是使用kafka进行数据流转，数据处理模块中包含多个微服务，单条数据会经历多次处理，部分业务耗时较长，导致在高频率接收到数据时候单体服务无法达到吞吐平衡，于是对于这些服务进行了分布式部署，多个实例进行消费处理。

业务实现

不指定分区

我们在给kafka发送消息时候，如果不指定分区，是不需要手动创建topic的,发送时没有topic,kafka会自动创建一个分区为1的topic，如下：

@Service
public class ProductService {

@Autowired
   private KafkaTemplate kafkaTemplate;

public void send(String msg, String topic) {
       kafkaTemplate.send(topic, msg);
   }
}

指定分区

topic分区初始化及配置

指定分区发送时候，如果未配置topic分区数，指定>0的分区，会提示分区不存在，这时候我们就需要提前创建好topic及分区

手动创建，服务启动前，使用kafka tool手动创建topic 不推荐 x

自动创建，服务启动时，使用KafkaClient创建 推荐 √

/**
* 初始化多分区的topic 基于springboot2
*/
@Component
public void TopicInitRunner implements ApplicationRunner {

@Autowired
   private AdminClient adminClient;

@Override
   public void run(ApplicationArguments args) throws Exception {
       // 通过配置文件读取自定义配置的topic名及分区数 省略...
       // Key topic V 分区数
       Map<String, Integer> topicPartitionMap = new HashMap<>();
       for (Map.Entry<String, Integer> e : topicPartitionMap.entrySet()) {
           createTopic(e.getKey(), e.getValue());
       }

}

public void createTopic(String topic, int partition) {
       NewTopic newTopic = new NewTopic(topic, partition);
       adminClient.createTopics(Lists.newArrayList(newTopic));
   }
}

/**
* 配置类参考 基于springboot2
* 如果只进行普通的单消息发送 无需添加此配置到项目中
*/
@Configuration
public class KafkaConfig {

@Value("${spring.kafka.bootstrap-servers}")
   private String servers;

@Bean
   public AdminClient adminClient() {
       return AdminClient.create(kafkaAdmin().getConfig());
   }

@Bean
   public KafkaAdmin kafkaAdmin() {
       Map<String, Object> props = Maps.newHashMap();
       props.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, servers);
       return new KafkaAdmin(props);
   }
}

生产者分区发送方案

上面讲到如何初始化分区topic，这时候我们的kafka环境已经准备好了，我们先使用TopicInitRunner为我们创建一个名称为 partition-topic 分区数为三，现在讲一讲如何均匀的讲消息发送的每个分区上，如何保证多消费者实例是负载均衡的，具体方案如下：

• 1.因为每条消息都是设备上传的，都会有设备id，先给每个设备生成一个自增号，这样1000个设备，每个设备就会有0到999的自增号，放到缓存中，每次根据消息中的设备id获取到该设备的自增号

• 2.使用自增号对分区数进行取模操作,代码实现如下：

public class ProductService {
   /**
    * data为需要发送的数据
    */
   public void partitionSend(String topic, int partition, JSONObject data) {
        // 获取设备id
       String deviceId = data.getString("deviceId");
       // 获取自增数 如果是新设备会创建一个并放入缓存中
       int inc = getDeviceInc(deviceId);
       // 如果分区数为3 设备自增id为1 取模结果为1 就是发送到1分区 这样1000个设备就可以保证每个分区发送数据量是1000 / 3
       int targetPartition = Math.floorMod(inc, partition);
       // 分区发送时候 需要指定一个唯一k 可以使用uuid或者百度提供的雪花算法获取id 字符串即可
       kafkaTemplate.send(topic, partition, getUuid(), data.toJSONString());
   }
}

消费者

我们讲到消费者使用分布式部署，一个微服务有多个实例，我们只需要按照服务监听的topic分区数创建对应数目的服务实例即可，这样kafka就会自动分配对应分区的数据到每个实例。

我们采取批量消费，进一步提高服务吞吐性能，消费及配置代码如下，配置文件参考springbootkafka配置即可，主要设计kafka服务配置，消费及生产配置，比较核心的是

@Component
public class DataListener {

@Autowired
   private MongoTemplate mongoTemplate;

/**
    * 站点报文监听消费
    *
    * @param records
    */
   @KafkaListener(topics = "partition-topic", containerFactory = "batchConsumerFactory")
   public void iotSiteHistoryMessageConsumer(List<ConsumerRecord<String, String>> records) {        
   }

/**
    * 消费者配置
    */
   @Bean
   public Map<String, Object> consumerConfigs() {
       Map<String, Object> props = Maps.newHashMap();
       props.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);
       props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
       props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, maxPollRecords);
       props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, 120000);
       props.put(ConsumerConfig.REQUEST_TIMEOUT_MS_CONFIG, 180000);
       props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
       props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
       return props;
   }

/**
    * 批量消费配置
    */
   @Bean
   public KafkaListenerContainerFactory batchConsumerFactory() {
       ConcurrentKafkaListenerContainerFactory<Integer, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
       factory.setConsumerFactory(new DefaultKafkaConsumerFactory<>(consumerConfigs()));
       factory.setBatchListener(true);
       return factory;
   }
}

来源：https://juejin.cn/post/6995746569580445709