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

Question

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

这是我的课；

@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();
        }
    }

这是我在应用程序中的Kafka属性。yml；

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流，对于常规SpringKafka，您需要如下属性：

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

还有：为什么使用@Payload String Content而不是已经序列化的cr.getVaue（）？