spring kafka流使用不起作用的函数样式消费来自多个主题的消息

@Configuration
@EnableKafka
public class KafkaConfig {

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

    @Bean
    public ConsumerFactory<String, String> consumerFactory() {
        Map<String, Object> props = new HashMap<>();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        return new DefaultKafkaConsumerFactory<>(props);
    }

    @Bean
    public ConcurrentKafkaListenerContainerFactory<String, String> kafkaListenerContainerFactory() {
        ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(consumerFactory());
        factory.setConcurrency(3); // 设置并发消费者数量
        factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.MANUAL); // 设置消息确认模式为手动确认
        return factory;
    }
}

@Component
public class KafkaMessageListener {

    @KafkaListener(topics = {"topic1", "topic2"})
    public void listen(ConsumerRecord<String, String> record) {
        // 处理接收到的消息
        System.out.println("Received message: " + record.value());
    }
}