Spring Cloud@StreamListener条件已弃用替代方案是什么

Question

我们有多个应用程序消费者收听同一个Kafka主题，生产者在向主题发送消息时设置消息头，以便特定实例可以评估消息头并处理消息。eg公司

@StreamListener(target=ITestSink.CHANNEL_NAME,condition="headers['franchiseName'] == 'sydney'")
public void fullfillOrder(@Payload TestObj message) {
    log.info("sydney order request received message is {}",message.getName());
}

在Spring Cloud Stream 3.0.0中，不推荐使用@StreamListener，我在函数中找不到与condition属性等效的属性。

有什么建议吗？

Answer 1

虽然我也找不到功能方法的等效方法，但我有一个建议。

StreamListener注释条件并不能阻止应用程序在将消息传递给侦听器之前必须使用消息、读取消息头并过滤掉特定记录这一事实（fullfillOrder（））。因此，可以安全地假设您正在使用与主题相关的所有消息（由Spring Cloud在后台为我们实现的事件接收器），但侦听器仅在header==sydney时执行。

如果有一种方法可以配置Spring Cloud使用的事件接收器（在命中侦听器之前丢弃消息），我建议对此进行研究。如果没有，将在进行任何处理之前过滤掉任何消息（非悉尼消息）。如果您熟悉Spring Cloud的功能性方法，可以这样看：

@Bean
public Consumer<Message<TestObj>> fulfillOrder() {
    return msg -> {
        // to get header - msg.getHeaders().get(key, valueType);
        // filter out bad messages
    }
}

或

@Bean
public Consumer<ConsumerRecord<?, TestObj>> fulfillOrder() {
    return msg -> {
        // msg.headers().lastHeader("franchiseName").value() -> filter em out
    }
}

其他：^my代码假设您正在通过spring-cloud-stream-binder-kafka将kafka客户端API与Spring Cloud stream集成。根据列出的标签，我将注意到Spring Cloud Stream有两个版本的Kafka绑定器-一个用于kafka客户端库，一个用于kafka流库。

在不考虑Spring云/框架的情况下，kafka streams中的高级别DSL不允许您访问标头，但低级别处理器API允许您访问标头。从这个示例来看，您似乎在利用客户端绑定器，而不是Spring-Cloud-Stream绑定器kafka streams。我还没有看到使用低级处理器API的spring cloud stream kafka streams binder的实现，所以我不知道这是否是目的所在。