java - Spring 集成 - 带有 SimpleAsyncTaskExecutor 的任务执行器

Question

在我的应用程序中，我试图对服务进行异步调用，这样我就不必等待那个昂贵的调用。我正在使用 SimpleAsyncTaskExecutor 但没有运气。请参阅下面我的代码片段。知道我错过了什么吗？

我的要求:请求到达主 channel ，即 issuTicket，它将检查 header 值并可以转到 flow1_channel 或 flow2_channel。 flow2 是一个昂贵的调用，所以我想开始一个新线程。

我尝试使用 SimpleAsyncTaskExecutor ，认为它会进行异步调用，但从日志来看，主线程仍在等待 flow2_channel 完成。有任何想法吗？预先感谢您对此的关注

 Config file :
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
   xmlns:int="http://www.springframework.org/schema/integration" xmlns:oxm="http://www.springframework.org/schema/oxm"
   xmlns:int-xml="http://www.springframework.org/schema/integration/xml" xmlns:int-ws="http://www.springframework.org/schema/integration/ws"
   xmlns:c="http://www.springframework.org/schema/c" xmlns:util="http://www.springframework.org/schema/util"
   xmlns:int-http="http://www.springframework.org/schema/integration/http" xmlns:task="http://www.springframework.org/schema/task"
   xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
              http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration-3.0.xsd
              http://www.springframework.org/schema/integration/http http://www.springframework.org/schema/integration/http/spring-integration-http-3.0.xsd
              http://www.springframework.org/schema/integration/xml http://www.springframework.org/schema/integration/xml/spring-integration-xml-3.0.xsd
              http://www.springframework.org/schema/integration/ws http://www.springframework.org/schema/integration/ws/spring-integration-ws-3.0.xsd
              http://www.springframework.org/schema/oxm http://www.springframework.org/schema/oxm/spring-oxm.xsd
              http://www.springframework.org/schema/util http://www.springframework.org/schema/util/spring-util.xsd
               http://www.springframework.org/schema/integration/http http://www.springframework.org/schema/integration/http/spring-integration-http-3.0.xsd
               http://www.springframework.org/schema/task http://www.springframework.org/schema/task/spring-task-3.0.xsd">

   <!-- Channel definition -->
   <int:channel id="flow2_channel">
      <int:dispatcher task-executor="asynchThread" />
   </int:channel>

   <int:channel id="flow1_channel" />

   <bean id="asynchThread" class="org.springframework.core.task.SimpleAsyncTaskExecutor">
      <property name="concurrencyLimit" value="20" />
   </bean>

   <!-- Main entry -->
   <int:chain input-channel="issueTicket">
      <int:filter expression="headers['status'] == 'TRUE'" discard-channel="flow1_channel" />
      <int:gateway request-channel="flow2_channel" />
   </int:chain>

   <int:chain input-channel="flow2_channel">
      <!-- time consuming call -->
      <int-http:outbound-gateway url="http://localhost:8080" http-method="POST" />
   </int:chain>

</beans>


  Unit test :
package com.swacorp.ais.createBooking.springorchestration;

import org.junit.Before;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.integration.MessageChannel;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;

@ContextConfiguration(locations = {"classpath:createBooking/TestContext.xml"})
@RunWith(SpringJUnit4ClassRunner.class)
public class SampleTest {

   @Autowired
   private MessageChannel issueTicket;

   @Before
   public void setUp() throws Exception {
   }

   @Test
   public void test() {
      issueTicket.send(MessageBuilder.withPayload("test").setHeader("status", "TRUE").build());
   }

}


---------------
From logs:
2015-09-04 13:29:51,029 DEBUG [main] {} [org.springframework.integration.handler.MessageHandlerChain]- org.springframework.integration.handler.MessageHandlerChain#0 received message: [Payload=test][Headers={timestamp=1441391391029, id=5f7c6af9-5efa-4cca-a274-2fa35097e10d, status=TRUE}]
2015-09-04 13:29:51,029 DEBUG [main] {} [org.springframework.integration.filter.MessageFilter]- org.springframework.integration.filter.MessageFilter@4c71eba0 received message: [Payload=test][Headers={timestamp=1441391391029, id=5f7c6af9-5efa-4cca-a274-2fa35097e10d, status=TRUE}]
2015-09-04 13:29:51,029 DEBUG [main] {} [org.springframework.beans.factory.support.DefaultListableBeanFactory]- Returning cached instance of singleton bean 'integrationEvaluationContext'
2015-09-04 13:29:51,029 DEBUG [main] {} [org.springframework.integration.filter.MessageFilter]- handler 'org.springframework.integration.filter.MessageFilter@4c71eba0' sending reply Message: [Payload=test][Headers={timestamp=1441391391029, id=5f7c6af9-5efa-4cca-a274-2fa35097e10d, status=TRUE}]
2015-09-04 13:29:51,029 DEBUG [main] {} [org.springframework.integration.gateway.RequestReplyMessageHandlerAdapter]- (inner bean)#1 received message: [Payload=test][Headers={timestamp=1441391391029, id=5f7c6af9-5efa-4cca-a274-2fa35097e10d, status=TRUE}]
2015-09-04 13:29:51,029 DEBUG [main] {} [org.springframework.beans.factory.support.DefaultListableBeanFactory]- Returning cached instance of singleton bean 'integrationEvaluationContext'
2015-09-04 13:29:51,029 DEBUG [main] {} [org.springframework.integration.channel.ExecutorChannel]- preSend on channel 'flow2_channel', message: [Payload=test][Headers={timestamp=1441391391029, id=b8ec3256-13f0-b86f-f57c-e021e0541dd4, errorChannel=org.springframework.integration.core.MessagingTemplate$TemporaryReplyChannel@2aed913b, replyChannel=org.springframework.integration.core.MessagingTemplate$TemporaryReplyChannel@2aed913b, status=TRUE}]
2015-09-04 13:29:51,044 DEBUG [main] {} [org.springframework.core.task.SimpleAsyncTaskExecutor$ConcurrencyThrottleAdapter]- Entering throttle at concurrency count 0
2015-09-04 13:29:51,044 DEBUG [main] {} [org.springframework.integration.channel.ExecutorChannel]- postSend (sent=true) on channel 'flow2_channel', message: [Payload=test][Headers={timestamp=1441391391029, id=b8ec3256-13f0-b86f-f57c-e021e0541dd4, errorChannel=org.springframework.integration.core.MessagingTemplate$TemporaryReplyChannel@2aed913b, replyChannel=org.springframework.integration.core.MessagingTemplate$TemporaryReplyChannel@2aed913b, status=TRUE}]
2015-09-04 13:29:51,044 DEBUG [SimpleAsyncTaskExecutor-1] {} [org.springframework.integration.handler.MessageHandlerChain]- org.springframework.integration.handler.MessageHandlerChain#1 received message: [Payload=test][Headers={timestamp=1441391391029, id=b8ec3256-13f0-b86f-f57c-e021e0541dd4, errorChannel=org.springframework.integration.core.MessagingTemplate$TemporaryReplyChannel@2aed913b, replyChannel=org.springframework.integration.core.MessagingTemplate$TemporaryReplyChannel@2aed913b, status=TRUE}]
2015-09-04 13:29:51,044 DEBUG [SimpleAsyncTaskExecutor-1] {} [org.springframework.integration.http.outbound.HttpRequestExecutingMessageHandler]- org.springframework.integration.handler.MessageHandlerChain#1$child#0.handler received message: [Payload=test][Headers={timestamp=1441391391029, id=b8ec3256-13f0-b86f-f57c-e021e0541dd4, errorChannel=org.springframework.integration.core.MessagingTemplate$TemporaryReplyChannel@2aed913b, replyChannel=org.springframework.integration.core.MessagingTemplate$TemporaryReplyChannel@2aed913b, status=TRUE}]
2015-09-04 13:29:51,044 DEBUG [SimpleAsyncTaskExecutor-1] {} [org.springframework.integration.http.support.DefaultHttpHeaderMapper]- outboundHeaderNames=[Accept, Accept-Charset, Accept-Encoding, Accept-Language, Accept-Ranges, Authorization, Cache-Control, Connection, Content-Length, Content-Type, Cookie, Date, Expect, From, Host, If-Match, If-Modified-Since, If-None-Match, If-Range, If-Unmodified-Since, Max-Forwards, Pragma, Proxy-Authorization, Range, Referer, TE, Upgrade, User-Agent, Via, Warning]
2015-09-04 13:29:51,044 DEBUG [SimpleAsyncTaskExecutor-1] {} [org.springframework.integration.http.support.DefaultHttpHeaderMapper]- headerName=[timestamp] WILL NOT be mapped
2015-09-04 13:29:51,044 DEBUG [SimpleAsyncTaskExecutor-1] {} [org.springframework.integration.http.support.DefaultHttpHeaderMapper]- headerName=[id] WILL NOT be mapped
2015-09-04 13:29:51,044 DEBUG [SimpleAsyncTaskExecutor-1] {} [org.springframework.integration.http.support.DefaultHttpHeaderMapper]- headerName=[errorChannel] WILL NOT be mapped
2015-09-04 13:29:51,044 DEBUG [SimpleAsyncTaskExecutor-1] {} [org.springframework.integration.http.support.DefaultHttpHeaderMapper]- headerName=[replyChannel] WILL NOT be mapped
2015-09-04 13:29:51,044 DEBUG [SimpleAsyncTaskExecutor-1] {} [org.springframework.integration.http.support.DefaultHttpHeaderMapper]- headerName=[status] WILL NOT be mapped
2015-09-04 13:29:51,060 DEBUG [SimpleAsyncTaskExecutor-1] {} [org.springframework.web.client.RestTemplate]- Created POST request for "http://localhost:8080"
2015-09-04 13:29:51,060 DEBUG [SimpleAsyncTaskExecutor-1] {} [org.springframework.web.client.RestTemplate]- Writing [test] as "text/plain;charset=UTF-8" using [org.springframework.http.converter.StringHttpMessageConverter@688d9b8]
2015-09-04 13:29:52,076 DEBUG [SimpleAsyncTaskExecutor-1] {} [org.springframework.beans.factory.support.DefaultListableBeanFactory]- Returning cached instance of singleton bean 'errorChannel'
2015-09-04 13:29:52,076 DEBUG [SimpleAsyncTaskExecutor-1] {} [org.springframework.core.task.SimpleAsyncTaskExecutor$ConcurrencyThrottleAdapter]- Returning from throttle at concurrency count 0
2015-09-04 13:29:52,076 DEBUG [main] {} [org.springframework.test.context.support.DirtiesContextTestExecutionListener]- After test method: context [[TestContext@6f648f32 testClass = SampleTest, testInstance = com.swacorp.ais.createBooking.springorchestration.SampleTest@32f554c0, testMethod = test@SampleTest, testException = org.springframework.integration.MessageHandlingException: error occurred in message handler [(inner bean)#1], mergedContextConfiguration = [MergedContextConfiguration@39cb7ee5 testClass = SampleTest, locations = '{classpath:createBooking/TestContext.xml}', classes = '{}', contextInitializerClasses = '[]', activeProfiles = '{}', contextLoader = 'org.springframework.test.context.support.DelegatingSmartContextLoader', parent = [null]]]], class dirties context [false], class mode [null], method dirties context [false].
2015-09-04 13:29:52,076 DEBUG [main] {} [org.springframework.test.context.web.ServletTestExecutionListener]- Resetting RequestContextHolder for test context [TestContext@6f648f32 testClass = SampleTest, testInstance = com.swacorp.ais.createBooking.springorchestration.SampleTest@32f554c0, testMethod = test@SampleTest, testException = org.springframework.integration.MessageHandlingException: error occurred in message handler [(inner bean)#1], mergedContextConfiguration = [MergedContextConfiguration@39cb7ee5 testClass = SampleTest, locations = '{classpath:createBooking/TestContext.xml}', classes = '{}', contextInitializerClasses = '[]', activeProfiles = '{}', contextLoader = 'org.springframework.test.context.support.DelegatingSmartContextLoader', parent = [null]]].
2015-09-04 13:29:52,076 DEBUG [main] {} [org.springframework.test.context.support.DirtiesContextTestExecutionListener]- After test class: context [[TestContext@6f648f32 testClass = SampleTest, testInstance = [null], testMethod = [null], testException = [null], mergedContextConfiguration = [MergedContextConfiguration@39cb7ee5 testClass = SampleTest, locations = '{classpath:createBooking/TestContext.xml}', classes = '{}', contextInitializerClasses = '[]', activeProfiles = '{}', contextLoader = 'org.springframework.test.context.support.DelegatingSmartContextLoader', parent = [null]]]], dirtiesContext [false].
2015-09-04 13:29:52,076 INFO [Thread-0] {} [org.springframework.context.support.GenericApplicationContext]- Closing org.springframework.context.support.GenericApplicationContext@2b29f93b: startup date [Fri Sep 04 13:29:50 CDT 2015]; root of context hierarchy
2015-09-04 13:29:52,076 DEBUG [Thread-0] {} [org.springframework.beans.factory.support.DefaultListableBeanFactory]- Returning cached instance of singleton bean 'org.springframework.integration.config.IdGeneratorConfigurer#0'

问题:为什么线程不并行运行？我可以看到主线程，然后是 SimpleAsynchTaskExecutor，然后主线程继续......我需要的是main、SimpleAsynchTaskExecutor、main、SimpleAsynchTaskExecutor 等..

Answer 1

什么是flow2_channel的“上游” ？

当询问此类问题时，最好显示您的所有配置。

如果它是某种网关，您需要设置 reply-timeout="0" (或 default-reply-timeout 为简单的 <gateway/> )如果您想忽略回复。

编辑

由于您使用的是中流网关，如果您不想等待结果，则需要对其进行配置...

<int:gateway request-channel="flow2_channel" default-reply-timeout="0" />

这就是我在第一个答案中所说的。

也就是说，目前还不清楚为什么你在那里有一个网关；您可以简单地将两条链与异步 channel 连接在一起...

<int:chain input-channel="issueTicket" output-channel="flow2_channel">
    <int:filter expression="headers['status'] == 'TRUE'" discard-channel="flow1_channel" />
</int:chain>

<int:chain input-channel="flow2_channel" output-channel="nullChannel">
    <!-- time consuming call -->
    <int-http:outbound-gateway url="http://localhost:8080" http-method="POST" />
</int:chain>