使用新版本 Chelsea.RC1 的 "condition paramter header" @StreamListener 时出错
Error using "condition paramter header" @StreamListener of new release Chelsea.RC1
我正在尝试使用事件过滤器来减少应用程序使用的主题数量,使用 spring 云流 (Chelsea.RC1) 的新版本中可用的新功能。正在创建消息,header 正确,但是,检查 queue 中的消息内容,消息不包含 header,仅包含 body与有效载荷。
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
您必须映射您的自定义 headers:
spring.cloud.stream.kafka.binder.headers = brand,operation
该信息存在于 documentation。
我正在尝试使用事件过滤器来减少应用程序使用的主题数量,使用 spring 云流 (Chelsea.RC1) 的新版本中可用的新功能。正在创建消息,header 正确,但是,检查 queue 中的消息内容,消息不包含 header,仅包含 body与有效载荷。
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
您必须映射您的自定义 headers:
spring.cloud.stream.kafka.binder.headers = brand,operation
该信息存在于 documentation。