15

我有一个严重依赖 Spring Integration 的自执行 jar 程序。我遇到的问题是程序在其他 Spring bean 完全完成之前终止。

下面是我正在使用的代码的精简版本，如果需要，我可以提供更多代码/配置。入口点是一个 main() 方法，它引导 Spring 并启动导入过程：

public static void main(String[] args) {
    ctx = new ClassPathXmlApplicationContext("flow.xml");
    DataImporter importer = (DataImporter)ctx.getBean("MyImporterBean");
    try {
        importer.startImport();
    } catch (Exception e) {
        e.printStackTrace();
    } finally {
        ctx.close();
    }
}

DataImporter 包含一个简单的循环，将消息发送到 Spring Integration 网关。这为流提供了一种主动的“推送”方法，而不是轮询数据的常用方法。这就是我的问题所在：

public void startImport() throws Exception {
    for (Item item : items) {
        gatewayBean.publish(item);
        Thread.sleep(200); // Yield period
    }
}

为了完整起见，流 XML 看起来像这样：

<gateway default-request-channel="inChannel" service-interface="GatewayBean" />

<splitter input-channel="inChannel" output-channel="splitChannel" />

<payload-type-router input-channel="splitChannel">
    <mapping type="Item" channel="itemChannel" />
    <mapping type="SomeOtherItem" channel="anotherChannel" />
</payload-type-router>

<outbound-channel-adapter channel="itemChannel" ref="DAOBean" method="persist" />

流程有效地启动和处理项目，但是一旦 startImport() 循环完成，主线程就会终止并立即关闭所有 Spring Integration 线程。这会导致竞争条件，最后 (n) 项在程序终止时未完全处理。

我想维护我正在处理的项目的引用计数，但这被证明是相当复杂的，因为流程经常将消息拆分/路由到多个服务激活器 - 这意味着很难确定每个项目是否具有“完成的”。

我认为我需要一些方法来检查是否没有 Spring bean 仍在执行，或者在终止之前标记发送到网关的所有项目都已完全处理。

我的问题是，我该如何做这些，或者有没有更好的方法来解决我没有想到的问题？

4

2 回答 2

10

您在这里没有使用请求-响应模式。

outbound-channel-adapter 是一个fire and forget action，如果你想等待响应，你应该使用一个等待响应的出站网关，并将响应连接到原始网关，然后在 java sendAndReceive 中不仅仅是发布。

于 2012-01-19T15:19:23.833 回答
3

如果您Item可以确定是否仍然需要它（processingFinished() 或在后端阶段执行的类似操作），您可以Item在中央机构注册所有 s，该机构跟踪非-finished Items 并有效地确定终止条件。

如果这种方法可行，您甚至可以考虑将项目打包成FutureTask-objects 或使用java.util.concurrent.

编辑：第二个想法：

你有没有想过让频道更智能？一旦不再发送任何数据，发送者就会关闭通道。在这种情况下，worker-bean 不必是守护线程，但可以根据封闭且空的输入通道确定其终止标准。

于 2012-01-19T12:04:33.963 回答