无法确定,我的Kafka侦听器配置有什么问题。最初,我有一个名为“transactions”的非空kafka主题,其中有几个记录(我可以在kafkatool中看到)。这是application.yml:
spring:
###
# Kafka Settings
###
kafka:
consumer:
bootstrap-servers: localhost:9092
key-deserializer: com.panbet.externalbet.history.report.support.ReportsBetKeyJsonDeserializer
value-deserializer: com.panbet.externalbet.history.report.support.ReportsBetJsonDeserializer
group-id: external.history.group
以下是java配置文件:
@EnableKafka
@Configuration
public class KafkaConfig
{
private final KafkaProperties properties;
public KafkaConfig(KafkaProperties properties)
{
this.properties = properties;
}
@Bean
public ConsumerFactory<ReportsBetKeyDto, ReportsBetDto> kafkaConsumerFactory()
{
return new DefaultKafkaConsumerFactory<>(properties.buildConsumerProperties());
}
@Bean
public ConcurrentKafkaListenerContainerFactory<ReportsBetKeyDto, ReportsBetDto> kafkaListenerContainerFactory()
{
ConcurrentKafkaListenerContainerFactory<ReportsBetKeyDto, ReportsBetDto> factory = new ConcurrentKafkaListenerContainerFactory<>();
factory.setConsumerFactory(kafkaConsumerFactory());
return factory;
}
}
我的Kafka监听器看起来像(前面提到的主题“事务”的监听器):
@Component
public class ReportsConsumer
{
@KafkaListener(topics = { "transactions" })
public void listen(ConsumerRecord<ReportsBetKeyDto, ReportsBetDto> record)
{
System.out.println(record);
}
}
我期望:当应用程序启动时,我会在reportsconsumer.listen方法中捕获debudder。但毫无收获的是,什么也没发生。听不到Kafka的主题。有什么问题吗?谢谢您。
1条答案
按热度按时间eagi6jfj1#
默认情况下,新消费者从主题末尾开始消费。
套
spring.kafka.consumer.auto-offset-reset=earliest
我也建议spring.kafka.consumer.enable-auto-commit=false
以便容器管理偏移量。