@KafkaListener 获取来自特定 kafka 主题的所有消息

Question

我有一个 @KafkaListener 方法来获取主题中的所有消息，但是对于 @Scheduled 方法起作用的每个间隔时间，我只得到一条消息。如何一次获取主题的所有消息？

这是我的课；

@Slf4j
@Service
public class KafkaConsumerServiceImpl implements KafkaConsumerService {

    @Autowired
    private SimpMessagingTemplate webSocket;

    @Autowired
    private KafkaListenerEndpointRegistry registry;

    @Autowired
    private BrokerProducerService brokerProducerService;

    @Autowired
    private GlobalConfig globalConfig;

    @Override
    @KafkaListener(id = "snapshotOfOutagesId", topics = Constants.KAFKA_TOPIC, groupId = "snapshotOfOutages", autoStartup = "false")
    public void consumeToSnapshot(ConsumerRecord<String, OutageDTO> cr, @Payload String content) {
        log.info("Received content from Kafka notification to notification-snapshot topic: {}", content);
        MessageListenerContainer listenerContainer = registry.getListenerContainer("snapshotOfOutagesId");
        JSONObject jsonObject= new JSONObject(content);
        Map<String, Object> outageMap = jsonToMap(jsonObject);
        brokerProducerService.sendMessage(globalConfig.getTopicProperties().getSnapshotTopicName(),
                outageMap.get("outageId").toString(), toJson(outageMap));
        listenerContainer.stop();
    }

    @Scheduled(initialDelayString = "${scheduler.kafka.snapshot.monitoring}",fixedRateString = "${scheduler.kafka.snapshot.monitoring}")
    private void consumeWithScheduler() {
        MessageListenerContainer listenerContainer = registry.getListenerContainer("snapshotOfOutagesId");
        if (listenerContainer != null){
            listenerContainer.start();
        }
    }

这是我在 application.yml 中的 kafka 属性；

kafka:
  streams:
    common:
      configs:
        "[bootstrap.servers]": 192.168.99.100:9092
        "[client.id]": event
        "[producer.id]": event-producer
        "[max.poll.interval.ms]": 300000
        "[group.max.session.timeout.ms]": 300000
        "[session.timeout.ms]": 200000
        "[auto.commit.interval.ms]": 1000
        "[auto.offset.reset]": latest
        "[group.id]": event-consumer-group
        "[max.poll.records]": 1

还有我的 KafkaConfiguration 类；

    @Bean
    public Map<String, Object> consumerConfigs() {
        Map<String, Object> props = new HashMap<>(globalConfig.getBrokerProperties().getConfigs());
        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "latest");
        return props;
    }

    @Bean
    public ConsumerFactory<String, String> consumerFactory() {
        return new DefaultKafkaConsumerFactory<>(consumerConfigs(), new StringDeserializer(), new StringDeserializer());
    }

    @Bean
    public KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<String, String>> kafkaListenerContainerFactory() {
        ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(consumerFactory());
        return factory;
    }

Answer 1

你目前正在做的是：

1. 创建一个侦听器，但尚未启动它 (autoStartup = false)
2. 当计划的作业启动时，启动容器（将开始使用来自主题的第一条消息）
3. 当第一条消息被消费时，您停止容器（导致不再消费任何消息）

因此，您所描述的行为确实不足为奇。

@KafkaListener 不需要计划任务即可开始使用消息。我想你可以删除autoStartup = false并删除预定的作业，之后监听器将一一消费该主题的所有消息，并等待新的消息出现在该主题上。

另外，我注意到的其他一些事情：

这些属性适用于 Kafka Streams，对于常规 Spring Kafka，您需要如下属性：

spring:
  kafka:
    bootstrap-servers: localhost:9092
    consumer:
      auto-offset-reset: earliest
      ...etc

另外：为什么要使用@Payload String content 而不是已经序列化的cr.getVaue()？