2

我正在尝试并行处理 SQL 查询的输出。下面给出的是我的代码。我在聚合器中有 sysout。但是我随机看到 Aggregator 中的 sysout 没有被打印出来。此外，聚合器中的发布方法也不是打印系统输出。我想，我在某些地方丢失了消息。任何人都可以解释一下吗？

    <int:bridge input-channel="inputChannel" output-channel="dbRequestChannel" />

        <jdbc:outbound-gateway request-channel="dbRequestChannel"
            max-rows-per-poll="0" data-source="dataSource" reply-channel="headerEnricher"
            query="select empname, empno, empdob from employee where empno = 1234" />

        <int:header-enricher input-channel="headerEnricher"
            output-channel="splitterChannel">
            <int:header name="payloadSize" value="3"></int:header>
        </int:header-enricher>

        <int:chain input-channel="splitterChannel" output-channel="splitterOutputChannel">
            <int:splitter />
        </int:chain>


        <int:channel id="splitterOutputChannel">
            <int:dispatcher task-executor="sampleTaskExecutor" />
        </int:channel>

        <bean id="sampleTaskExecutor"
            class="org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor">
            <property name="corePoolSize" value="5" />
            <property name="maxPoolSize" value="10" />
            <property name="queueCapacity" value="25" />
        </bean>

        <int:service-activator input-channel="splitterOutputChannel"
            ref="springIntegrationtest" method="testMethod" output-channel="aggregatePayload">
        </int:service-activator>

        <int:aggregator input-channel="aggregatePayload"
            release-strategy-method="release" output-channel="nullChannel"
            send-partial-result-on-expiry="true" ref="springIntegrationtest"
            method="aggregateData" />

    @RunWith(SpringJUnit4ClassRunner.class)
@ContextConfiguration(locations = { "classpath:spring-integration.xml" })
public class SpringIntegrationTest {


    @Autowired
    private MessageChannel inputChannel;

    SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SSS");



    @Test
    public void testQueue() {
        Message<String> quoteMessage = MessageBuilder
                .withPayload("testPayload").build();
        inputChannel.send(quoteMessage);
    }


    public Map<String, String> testMethod(Message<?> m) {

        System.out.println(sdf.format(new Date()));
        return (Map<String, String>) m.getPayload();
    }

    public boolean release(ArrayList<Map<String, Object>> payload) {
        boolean release = false;
        int size = payload.size();
        if (size == 3) {
            release = true;
        }
        System.out.println(release);
        return release;
    }

    public Message<String> aggregateData(ArrayList<Map<String, Object>> payload) {

        System.out.println("In aggregateData " + payload);
        Message<String> quoteMessage = MessageBuilder
                .withPayload("testPayload").build();
        return quoteMessage;
    }

}
4

1 回答 1

1

好吧，我认为您的问题在于状态和选项的组合。

你有这个：

int size = payload.size();
if (size == 3) {
    release = true;
}

因此，您的聚合器将在项目到达后立即释放组3，同时拆分后您可能会有更多项目。

信号聚合器release完成组并退出。默认情况下，它有一个类似 的选项expireGroupsUponCompletion = false，这意味着在商店中保留completed组但具有状态。

如果您的目标只是从聚合器元组中发出 3 个，您应该考虑expireGroupsUponCompletion将true. 有关详细信息，请参阅聚合器手册。

于 2016-07-19T00:46:48.110 回答