spring-cloud - 使用新版本 Chelsea.RC1 的 "condition paramter header"@StreamListener 时出错

标签 spring-cloud spring-cloud-stream spring-kafka

我正在尝试使用事件过滤器来减少应用程序使用的主题数量,使用新版本的 spring 云流 (Chelsea.RC1) 中可用的新功能。正在创建具有正确 header 的消息,但是,检查队列中消息的内容,消息不包含 header ,仅包含带有有效负载的正文。

public void sendEnroll(EnrollCommand data) {

    //MessageChannel
    outputEnroll.send(MessageBuilder
            .withPayload(data)
                .setHeader("brand", "MASTERCARD")
                .setHeader("operation", Operation.ENROLL).build());
}

消费者

@Service
@EnableBinding(Channel.class)
public class EnrollConsumer {

@Autowired
private EnrollService service;

@StreamListener(target = Channel.INPUT_ENROLL, condition = "headers['brand']=='MASTERCARD'")
public void enrollConsumer(@Payload String command){
    System.out.println(command);
    //service.enrollment(command);
 }
}

在消费者服务中,它给出如下警告:

WARN -kafka-listener-1 o.s.c.s.b.DispatchingStreamListenerMessageHandler:62 - Cannot find a @StreamListener matching for message with id: 7baae934-7484-a7fd-91b0-ba906558bb13

最佳答案

您必须映射您的自定义 header :

spring.cloud.stream.kafka.binder.headers = brand,operation

该信息存在于 documentation 中.

关于spring-cloud - 使用新版本 Chelsea.RC1 的 "condition paramter header"@StreamListener 时出错,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/42910265/

相关文章:

java - 如何在 Spring Cloud Gateway 中添加特定于路由的自定义过滤器

java - 使用Hystrix通过断路器从Eureka获取服务列表

java.lang.NoClassDefFoundError : org/springframework/cloud/test/TestSocketUtils after upgrading Spring Boot version to 3. x

Spring Netflix eureka、zuul vs Spring 云数据流

spring-boot - 停止 Spring Cloud Stream @StreamListener 监听,直到收到某些 Spring 事件

java - Jhipster 中的 Liquibase 迁移异常

java - 使用 KafkaListener 从 Kafka 消费一次最少 N 条消息

java - Feign客户端执行失败 : java. lang.reflect.InitationTargetException

spring-boot - Kafka ConsumerRecord 返回 null

JAAS/SASL 的 spring-kafka application.properties 配置不起作用