我创建了一个流,它使用来自 RabbitMQ 的消息,然后使用路由器按类型分发到相应的服务。服务中的方法采用参数Message<?>
,因为我需要在那里使用标头。但是在这种方法中,我只收到类型为 java.lang.String
的消息有效负载,而不是org.springframework.messaging.Message
和我收到错误java.lang.ClassCastException: java.lang.String cannot be cast to org.springframework.messaging.Message
.
有效负载不适合我,因为我需要从消息中获取标头。
@Bean
public IntegrationFlow testFlow(String queueName,
ConnectionFactory connectionFactory,
Service1 service1,
Service2 service2) {
SimpleMessageListenerContainer consumerListener = new SimpleMessageListenerContainer(connectionFactory);
consumerListener.addQueueNames(queueName);
return IntegrationFlows.from(Amqp.inboundAdapter(consumerListener))
.transform(s -> s, ConsumerEndpointSpec::transactional)
.<Message<?>, String>route(HeadersUtil::getType, m -> m
.subFlowMapping(Type.SERVICE_1, sf -> sf.handle(service1::handleProcedure))
.subFlowMapping(Type.SERVICE_2, sf -> sf.handle(service2::handleProcedure)))
.get();
}
方法handleProcedure
的签名如下:
void handleProcedure(Message<?> message)
我希望在方法handleProcedure
中获取Message
的标头,但现在我得到了例外。
我认为您没有正确理解堆栈跟踪。
您的void handleProcedure(Message<?> message)
及其service1::handleProcedure
方法引用完全符合IntegrationFlowDefinition
中的public B handle(MessageHandler messageHandler) {
方法签名。
您的问题就在这里:
.<Message<?>, String>route(HeadersUtil::getType,
您的HeadersUtil::getType
需要消息,但 lambda 调用的类型是payload
,在您的情况下String
。
这应该有效:
.<Message<?>, String>route(Message.class, HeadersUtil::getType,
您正在使用的 .handle
的handle(GenericHandler<P>)
版本仅获取有效负载。
如果要接收完整的消息,则需要使用不同的重载.handle
,例如handle("service1", "handleProcedure")
或.handle(service1, "handleProcedure")
。