java - 为什么 Spring Integration 消息方法处理行为在默认 Spring Cloud Stream 配置中发生了变化

Question

我有一个已经使用 Spring Integration(最新 5.1.6)的应用程序。配置如下流程:

@Configuration
public class SomeConfigClass {
...
    @MessagingGateway(name = "someGateway")
    interface Gateway {
        @Gateway(requestChannel = "inboundChannel")
        @Payload("T(java.time.ZonedDateTime).now()")
        void replicate();
    }

    @Bean
    public DirectChannel inboundChannel() {
        return MessageChannels.direct().get();
    }

    @Bean
    public IntegrationFlow someFlow() {
        return IntegrationFlows.from(inboundChannel())
                .handle(someHandler())
                .channel(OUT)
                .get();
    }

    @Bean
    public SomeHandler someHandler() {
        return new SomeHandler();
    }
}

和

public class SomeHandler implements GenericHandler<Object> {
    @Override
    public Message<List<String>> handle(final Object payload, 
                                        final MessageHeaders headers) {
        ...
        return MessageBuilder
                .withPayload(someList)
                .copyHeaders(headers)
                .setHeader("custom", customHeader)
                .build();
    }
}

一切正常。

如果我尝试在初始化的上下文中找到 integrationArgumentResolverMessageConverter bean，我会看到下一个转换器:

• MappingJackson2MessageConverter
• ByteArrayMessageConverter
• ObjectStringMessageConverter
• GenericMessageConverter

之后，我将 Spring Cloud Stream 2.1.2 依赖项和 Kinesis Binder 1.2.0 添加到我的 pom 依赖项中。默认配置绑定(bind)。

应用程序启动，但当我尝试处理现有流程时，它失败了，如下所示:

EL1004E: Method call: Method handle(java.time.ZonedDateTime,org.springframework.messaging.MessageHeaders) cannot be found on type p.a.c.k.a.g.e.SomeHandler at org.springframework.expression.spel.ast.MethodReference.findAccessorForMethod(MethodReference.java:225) at org.springframework.expression.spel.ast.MethodReference.getValueInternal(MethodReference.java:134) at org.springframework.expression.spel.ast.MethodReference.access$000(MethodReference.java:54) at org.springframework.expression.spel.ast.MethodReference$MethodValueRef.getValue(MethodReference.java:390) at org.springframework.expression.spel.ast.CompoundExpression.getValueInternal(CompoundExpression.java:90) at org.springframework.expression.spel.ast.SpelNodeImpl.getTypedValue(SpelNodeImpl.java:114) at org.springframework.expression.spel.standard.SpelExpression.getValue(SpelExpression.java:365) at org.springframework.integration.util.AbstractExpressionEvaluator.evaluateExpression(AbstractExpressionEvaluator.java:172) at org.springframework.integration.util.AbstractExpressionEvaluator.evaluateExpression(AbstractExpressionEvaluator.java:160) at org.springframework.integration.handler.support.MessagingMethodInvokerHelper.invokeExpression(MessagingMethodInvokerHelper.java:664) at org.springframework.integration.handler.support.MessagingMethodInvokerHelper.invokeHandlerMethod(MessagingMethodInvokerHelper.java:655) at org.springframework.integration.handler.support.MessagingMethodInvokerHelper.processInternal(MessagingMethodInvokerHelper.java:491) at org.springframework.integration.handler.support.MessagingMethodInvokerHelper.process(MessagingMethodInvokerHelper.java:362) at org.springframework.integration.handler.MethodInvokingMessageProcessor.processMessage(MethodInvokingMessageProcessor.java:106) at org.springframework.integration.handler.ServiceActivatingHandler.handleRequestMessage(ServiceActivatingHandler.java:93) at org.springframework.integration.handler.AbstractReplyProducingMessageHandler.handleMessageInternal(AbstractReplyProducingMessageHandler.java:123) at org.springframework.integration.handler.AbstractMessageHandler.handleMessage(AbstractMessageHandler.java:169) at org.springframework.integration.dispatcher.AbstractDispatcher.tryOptimizedDispatch(AbstractDispatcher.java:115) at org.springframework.integration.dispatcher.UnicastingDispatcher.doDispatch(UnicastingDispatcher.java:132) at org.springframework.integration.dispatcher.UnicastingDispatcher.dispatch(UnicastingDispatcher.java:105) at org.springframework.integration.channel.AbstractSubscribableChannel.doSend(AbstractSubscribableChannel.java:73) at org.springframework.integration.channel.AbstractMessageChannel.send(AbstractMessageChannel.java:453) at org.springframework.integration.channel.AbstractMessageChannel.send(AbstractMessageChannel.java:401) at org.springframework.messaging.core.GenericMessagingTemplate.doSend(GenericMessagingTemplate.java:187) at org.springframework.messaging.core.GenericMessagingTemplate.doSend(GenericMessagingTemplate.java:166) at org.springframework.messaging.core.GenericMessagingTemplate.doSend(GenericMessagingTemplate.java:47) at org.springframework.messaging.core.AbstractMessageSendingTemplate.send(AbstractMessageSendingTemplate.java:109) at org.springframework.messaging.core.AbstractMessageSendingTemplate.convertAndSend(AbstractMessageSendingTemplate.java:151) at org.springframework.messaging.core.AbstractMessageSendingTemplate.convertAndSend(AbstractMessageSendingTemplate.java:143) at org.springframework.integration.gateway.MessagingGatewaySupport.send(MessagingGatewaySupport.java:413) at org.springframework.integration.gateway.GatewayProxyFactoryBean.invokeGatewayMethod(GatewayProxyFactoryBean.java:533) at org.springframework.integration.gateway.GatewayProxyFactoryBean.doInvoke(GatewayProxyFactoryBean.java:473) at org.springframework.integration.gateway.GatewayProxyFactoryBean.invoke(GatewayProxyFactoryBean.java:463) at org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:186) at org.springframework.aop.framework.JdkDynamicAopProxy.invoke(JdkDynamicAopProxy.java:212) at com.sun.proxy.$Proxy444.replicate(Unknown Source)

当我尝试从初始化的上下文中获取相同的 integrationArgumentResolverMessageConverter bean 时，我看到下一个链:

• ApplicationJsonMessageMarshallingConverter
• TupleJsonMessageConverter
• ByteArrayMessageConverter
• ObjectStringMessageConverter
• JavaSerializationMessageConverter
• KryoMessageConverter
• JsonUnmarshallingConverter

并且没有GenericMessageConverter。据我了解，由于缺少此转换器，它无法转换(如果我错了，请纠正我)。

为什么当我只添加 Spring Cloud Stream 默认配置时行为会有所不同？或者如何针对特定流程定制使用转换器链？或者如何保留不同集成流程的消息对话行为？

<小时/>

更新:因此，当我研究 Spring Cloud Stream 时，它不仅重新定义了默认集成 MessageConverter，而且还重新定义了默认 HandlerMethodArgumentResolver >s，用于将方法参数与消息映射..

添加 Spring Cloud Stream 之前:

• HeaderMethodArgumentResolver
• HeadersMethodArgumentResolver
• 消息方法参数解析器
• PayloadExpressionArgumentResolver
• NullAwarePayloadArgumentResolver
• PayloadsArgumentResolver
• MapArgumentResolver
• PayloadArgumentResolver

添加Spring Cloud Stream后:

• SmartPayloadArgumentResolver
• SmartMessageMethodArgumentResolver
• HeaderMethodArgumentResolver
• HeadersMethodArgumentResolver
• PayloadExpressionArgumentResolver
• NullAwarePayloadArgumentResolver
• PayloadExpressionArgumentResolver
• PayloadsArgumentResolver
• MapArgumentResolver

有两个已弃用的 SmartPayloadArgumentResolver 和 SmartMessageMethodArgumentResolver，修复了从 byte[] 有效负载到 Object 的转换。但我不明白为什么有两个 PayloadExpressionArgumentResolver？..

主要问题是:为什么 Spring Cloud Stream 默认应用程序上下文会影响 Spring Integration 默认应用程序上下文，我之前认为 Stream 的解析器/转换器仅与与流目标 channel 链接的消息端点相关......

Answer 1

我不确定为什么 Stream 会删除该转换器(可能是一个错误，也许在那里打开一个 GitHub 问题)，但我相信您可以将其作为 @StreamMessageConverter @ 添加回来Bean 如流文档中所述。