java - Spring Task Executor 在 JUNIT 中创建了额外的线程

Question

我的 spring-config.xml 中有简单的任务执行器，并附加了几个链

      <channel id="inputChannel" />
    <task:executor id="threadPoolExecutor" pool-size="2" />
    <publish-subscribe-channel id="multiCastChannel"
        task-executor="threadPoolExecutor" />

    <chain input-channel="inputChannel"
        output-channel="multiCastChannel">
        <json-to-object-transformer
            type="com.company.integration.domain.DomainObject" />
        <service-activator ref="validator"
            method="validate" />
</chain>

<chain input-channel="multiCastChannel"
        output-channel="inventoryAdjustmentOutputChannelOne">
        <service-activator ref="adapterOne"
            method="buildOutputMessageOne" />
</chain>
<chain input-channel="multiCastChannel"
        output-channel="inventoryAdjustmentOutputChannelTwo">
        <service-activator ref="adapterTwo"
            method="buildOutputMessageTwo" />
</chain>

当消息发布到“inputChannel”并经过处理并发送到“multuCastChannel”时，会创建两个线程，而不会出现以下问题

threadPoolExecutor-1 threadPoolExecutor-2

并且这两个仅在每个输入消息中创建一次，这很好。 但是当我尝试使用 JUnit 进行相同测试时...每个“multiCastChannel”链都执行两次...意味着链中的服务激活器(adapterOne、adapterTwo)正在调用两次 每条链......有线......

知道为什么 JUnit 有这种行为吗？

下面是 Junit 代码

@RunWith(SpringJUnit4ClassRunner.class)
@ContextConfiguration(locations = { "classpath:configuration/spring-config.xml"})
@FixMethodOrder(MethodSorters.NAME_ASCENDING)
@PropertySource("classpath:application.properties")
@SuppressWarnings("unchecked")
@WebAppConfiguration
@DirtiesContext(classMode = DirtiesContext.ClassMode.AFTER_EACH_TEST_METHOD)
public class InventoryAdjustmentMessageTest {


    @Autowired
    private DirectChannel inputChannel;


    @Test
    public void testTaskShed()
            throws IOException, InterruptedException, JMSException {
        String validInput = setup("valid-message.txt");
        Message<String> inputMessage = TextMessageUtil.createNewGenericMessage(validInput);
        inputChannel.send(inputMessage);
        Thread.sleep(5000);

}

Spring 集成版本:4.1.6

添加应用程序配置信息:

@Import({ HarnessConfiguration.class, LocalConfiguration.class, MongoDbConfiguration.class, WebConfiguration.class,
        WebsphereMQJMSConfiguration.class })
@Configuration
@EnableWebMvc
@ComponentScan(basePackages = { "com.abc.inventory.adjustment.integration.service",
        "com.abc.inventory.adjustment.integration.domain" }, useDefaultFilters = false, excludeFilters = {
                @ComponentScan.Filter(type = FilterType.ASSIGNABLE_TYPE, value = ApplicationConfiguration.class) }, includeFilters = {
                        @ComponentScan.Filter(type = FilterType.ANNOTATION, value = { Controller.class,
                                Component.class }) })
@ImportResource({ "classpath:configuration/spring-config.xml", "classpath:configuration/spring-adapters.xml" })
@EnableMongoRepositories("com.abc.inventory.adjustment.integration.service.audit.repository")
@EnableAutoConfiguration
@EnableMongoAuditing
public class ApplicationConfiguration extends WebMvcConfigurerAdapter {
//

}

添加调试快照

-Tej

Answer 1

在这里您可以找到简单的测试用例来演示正确的行为:

<task:executor id="executor" pool-size="2"/>

<publish-subscribe-channel id="pubSubChannel" task-executor="executor" />

<service-activator input-channel="pubSubChannel" expression="T(System).out.println(payload)"/>

<service-activator input-channel="pubSubChannel" expression="T(System).out.println('foo: ' + payload)"/>

<小时/>

@Test
public void testPubSubChannel() throws InterruptedException {
    ConfigurableApplicationContext context = new ClassPathXmlApplicationContext("pubSubChannelConfig.xml", getClass());
    MessageChannel channel = (MessageChannel) context.getBean("pubSubChannel");
    for (int i = 0; i < 10; i++) {
        channel.send(new GenericMessage<Integer>(i));
    }
    Thread.sleep(10000);
    context.close();
}

结果如下:

foo: 0
0
foo: 1
1
2
foo: 2
3
foo: 3
4
foo: 4
5
foo: 5
6
foo: 6
7
foo: 7
8
foo: 8
9
foo: 9