kafka 按照主題消費 和 按照分區消費

一次項目想，多線程消費主題的中的數據，百度了一波之后，可以按分區進行單獨消費，記錄一下

首先，傳統按照主題消費：

@KafkaListener(topics = {Constants.KAFKA_TOPIC_PISHH}, containerFactory = "kafkaListenerMonitorFactory")  2     public void listenDataInfo(List<ConsumerRecord> records, Acknowledgment ack) {
        try {
            long beginTime = System.currentTimeMillis();
            batchList.clear();
            for (ConsumerRecord record : records) {
                count++;
                Optional<?> kafkaMessage = Optional.ofNullable(record.value());
                //獲取kafka消息
               logger.info(kafkaMessage.get().toString());
            }
        } catch (Exception e){
            e.printStackTrace();
        } finally {
            ack.acknowledge();//手動提交偏移量
        }
    }

 

按照主題分區消費：

@KafkaListener(id = "id0", topicPartitions = { @TopicPartition(topic = Constants.KAFKA_TOPIC_PISHH, partitions = { "0" }) }, containerFactory = "kafkaListenerMonitorFactory")
public void listenDataInfo0(List<ConsumerRecord> records, Acknowledgment ack) {  3 try {  4 long beginTime = System.currentTimeMillis();  5  batchList.clear();  6 for (ConsumerRecord record : records) {  7 count++;  8 Optional<?> kafkaMessage = Optional.ofNullable(record.value());  9 //獲取kafka消息 10  logger.info(kafkaMessage.get().toString()); 11  } 12 13 } catch (Exception e){ 14  e.printStackTrace(); 15 } finally { 16 ack.acknowledge();//手動提交偏移量 17  } 18 }

 

參考：https://blog.csdn.net/russle/article/details/81258590