如何通过代码在KafkaListener中设置主题名称？

Question

我正在为Kafka主题编写侦听器，并使用@kafkaListener注释来做同样的事情。到目前为止，我硬编码了主题名（topic1），效果很好。这是一个有效的代码：-

@Component
public class KafkaConsumer {

@Autowired
private KafkaProperties kafkaProps;


@Autowired
@Qualifier("CustomObjectMapper")
private ObjectMapper objectMapper;

@KafkaListener(topics = "topic1", containerFactory = "createPokafkaListenerContainerFactory")
public void CreatePoListener(PurchaseOrder po, Acknowledgment ack)
    throws JsonProcessingException {

    LOG.info("Received message for po create from kafka topic {} is {}",
        kafkaProps.getOmsCreateTopicName(), objectMapper
            .writerWithDefaultPrettyPrinter().writeValueAsString(po));

    ack.acknowledge();


}

}

现在，当我试图更改代码以从代码中获取主题名称时，它不起作用。代码，我试图从代码中获取值后，在堆栈溢出（如何将动态主题名称传递给@kafkalistener（主题从环境变量）是：-

@Component
public class KafkaConsumer {

@Autowired
private KafkaProperties kafkaProps;

public KafkaProperties getKafkaProps() {
    return kafkaProps;
}

@Autowired
@Qualifier("CustomObjectMapper")
private ObjectMapper objectMapper;

@KafkaListener(topics = "#{__listener.kafkaProps.getOmsCreateTopicName()}", containerFactory = "omsCreatePokafkaListenerContainerFactory")
    public void CreatePoListener(PurchaseOrder po, Acknowledgment ack)
    throws JsonProcessingException {

    LOG.info("Received message for po create from kafka topic {} is {}",
        kafkaProps.getOmsCreateTopicName(), objectMapper
            .writerWithDefaultPrettyPrinter().writeValueAsString(po));

    ack.acknowledge();


}

}

当我试图启动服务器时，它抛出错误：-

bean初始化失败；嵌套的异常是org。springframework。豆。工厂BeanExpressionException:表达式解析失败；嵌套的异常是org。springframework。表示斯佩尔。SpelEvaluationException:EL1008E:在类型为“org”的对象上找不到属性或字段“_listener”。springframework。豆。工厂配置。BeanExpressionContext’——可能不公开？

谁能帮帮我我做错了什么？

Answer 1

在消费者类别中，您需要进行以下更改：

@Autowired
public KafkaProperties kafkaProps;

@KafkaListener(topics = "#{kafkaConsumer.kafkaProps.getOmsCreateTopicName()}"

这对我很管用。

Answer 2

1.1.x不再受支持；如果无法升级到Spring Framework 4.3之外，则应升级到1.3.10，这是与Spring 4.3一起使用的最新版本；它有一个比1.1更简单、更可靠的线程模型。x、 多亏了KIP-62。

虽然您不能在1.3. x中使用__listener，但您可以使用SpEL表达式直接从KafkaPropertiesbean获取主题：

@KafkaListener(topics = "#{kafkaProps.omsCreateTopicName}" ...)