java - Spring集成和多线程，丢失了必需的 header

Question

我是 Spring 集成的新手，我想将一些数据导出到 6 个 CSV 文件中，我使用 Spring 集成，它在单线程中工作得很好，但是当我想实现多线程方法时，我收到错误消息“所需的 header 不可用:”，并且在 6 个文件中我只生成了 3 或 2 个文件。

<bean id="taskExecutor"
    class="org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor">
    <property name="threadFactory" ref="threadFactory" />
    <property name="corePoolSize" value="20" />
</bean>


<bean id="threadFactory"
    class="com.sqli.hybris.fredhoppersearch.utils.ExportToFredhopperThreadFactory"
    scope="tenant">
    <constructor-arg
        value="#{T(de.hybris.platform.core.Registry).getCurrentTenant()}" />
</bean>


<int:gateway id="indexerGateway" default-request-channel="indexElementChannel"
    service-interface="com.sqli.hybris.fredhoppersearch.si.IndexerGateway" />



<int:channel id="indexElementChannel">
    <int:dispatcher task-executor="taskExecutor" />
</int:channel>

<!-- <int:channel id="indexElementChannel" /> -->

<int:filter input-channel="indexElementChannel"
    output-channel="itemFilteredChannel" ref="dataFilter" method="accept" />

<int:channel id="itemFilteredChannel" />

<int:header-value-router input-channel="itemFilteredChannel"
    header-name="mode">
    <int:mapping value="FULL" channel="fullChannel" />
    <int:mapping value="INCREMENTAL" channel="incrementalChannel" />
    <int:mapping value="RUPTURE" channel="deleteChannel" />
</int:header-value-router>

<int:channel id="fullChannel" />
<int:channel id="incrementalChannel" />
<int:channel id="deleteChannel" />

<!-- Data Transformer -->

<int:transformer input-channel="fullChannel"
    output-channel="itemTransformedChannel" ref="fullTransformer" method="transform" />

<int:transformer input-channel="incrementalChannel"
    output-channel="itemTransformedChannel" ref="incrementalTransformer"
    method="transform" />

<int:transformer input-channel="deleteChannel"
    output-channel="itemTransformedChannel" ref="deleteTransformer"
    method="transform" />


<int:aggregator input-channel="itemTransformedChannel"
    method="aggregateData" output-channel="aggregatedChannel"
    release-strategy="releaseBean" release-strategy-method="canRelease"
    send-timeout="12000" send-partial-result-on-expiry="false"
    expire-groups-upon-completion="true">
    <bean class="com.sqli.hybris.fredhoppersearch.si.FredhopperDataAggregator"
        scope="tenant" />
</int:aggregator>

<int:channel id="aggregatedChannel">
</int:channel>

<int:channel id="itemTransformedChannel" />
<int:header-value-router input-channel="aggregatedChannel"
    header-name="mode">
    <int:mapping value="FULL" channel="fullExportChannel" />
    <int:mapping value="INCREMENTAL" channel="fullExportChannel" />
    <int:mapping value="RUPTURE" channel="fullExportChannel" />
</int:header-value-router>


<int:channel id="fullExportChannel" />
<int:service-activator input-channel="fullExportChannel"
    ref="fullIndexer" method="index" />

<bean id="dataFilter"
    class="com.sqli.hybris.fredhoppersearch.si.FredhopperDataFilter"
    scope="tenant" />

<alias name="defaultReleaseBean" alias="releaseBean" />
<bean id="defaultReleaseBean"
    class="com.sqli.hybris.fredhoppersearch.si.FredhopperReleaseStrategy"
    scope="tenant" />

但是当我尝试使其在多线程中运行时。

我有这些消息

org.springframework.integration.MessageHandlingException: java.lang.IllegalArgumentException: required header not available:  exporterConfig
    at org.springframework.integration.handler.MethodInvokingMessageProcessor.processMessage(MethodInvokingMessageProcessor.java:76)
    at org.springframework.integration.handler.ServiceActivatingHandler.handleRequestMessage(ServiceActivatingHandler.java:64)
    at org.springframework.integration.handler.AbstractReplyProducingMessageHandler.handleMessageInternal(AbstractReplyProducingMessageHandler.java:97)
    at org.springframework.integration.handler.AbstractMessageHandler.handleMessage(AbstractMessageHandler.java:73)
    at org.springframework.integration.dispatcher.UnicastingDispatcher.doDispatch(UnicastingDispatcher.java:114)
    at org.springframework.integration.dispatcher.UnicastingDispatcher.dispatch(UnicastingDispatcher.java:101)

我还尝试将 spring beans 的范围更改为原型(prototype)，但它不起作用，任何想法，

感谢您的帮助，

Answer 1

java.lang.IllegalArgumentException: required header not available: exporterConfig

此异常由 MessagingMethodInvokerHelper 引发，当您使用@Header时对于方法参数之一。在你的情况下是 ServiceActivatingHandler 。

所以，你的一些<int:service-activator>使用该注释从该方法参数的 header 中提取值，但您的 Message 没有该 header MessageHeaders 。我没有看到您将该 header 放入消息的位置。

或者只是将其标记为 @Header(required = false) ，或者尝试找出您没有提供 exporterConfig 的地方 header 。