如何在将来取消时在Callable内终止CXF webservice调用

编辑

这个问题到目前为止已经经过了几次迭代,所以请随意查看修订版,以查看有关历史和尝试过的事情的一些背景信息。

我将CompletionService与ExecutorService和Callable一起使用,通过CXF生成的代码在几个不同的Web服务上并发调用多个函数。。这些服务都为我在项目中使用的一组信息提供了不同的信息。但是,服务可能会在长时间内无法响应,而不会引发异常,从而延长对组合信息集的等待时间。

为了解决这个问题,我将同时运行所有服务调用,几分钟后,我想终止任何尚未完成的调用,最好从可调用的内部或通过抛出详细的异常来记录哪些调用尚未完成。

下面是一些高度简化的代码来说明我已经在做的事情:

private Callable<List<Feature>> getXXXFeatures(final WiwsPortType port, 
final String accessionCode) {
    return new Callable<List<Feature>>() {
        @Override
        public List<Feature> call() throws Exception {
            List<Feature> features = new ArrayList<Feature>();
            //getXXXFeatures are methods of the WS Proxy
            //that can take anywhere from second to never to return
            for (RawFeature raw : port.getXXXFeatures(accessionCode)) {
                Feature ft = convertFeature(raw);
                features.add(ft);
            }
            if (Thread.currentThread().isInterrupted())
                log.error("XXX was interrupted");
            return features;
        }
    };
}

以及并发启动WS调用的代码:

WiwsPortType port = new Wiws().getWiws();
List<Future<List<Feature>>> ftList = new ArrayList<Future<List<Feature>>>();
//Counting wrapper around CompletionService, 
    //so I could implement ccs.hasRemaining()
CountingCompletionService<List<Feature>> ccs = 
        new CountingCompletionService<List<Feature>>(threadpool);
ftList.add(ccs.submit(getXXXFeatures(port, accessionCode)));
ftList.add(ccs.submit(getYYYFeatures(port accessionCode)));
ftList.add(ccs.submit(getZZZFeatures(port, accessionCode)));

List<Feature> allFeatures = new ArrayList<Feature>();
while (ccs.hasRemaining()) {
            //Low for testing, eventually a little more lenient
    Future<List<Feature>> polled = ccs.poll(5, TimeUnit.SECONDS);
    if (polled != null)
        allFeatures.addAll(polled.get());
    else {
        //Still jobs remaining, but unresponsive: Cancel them all
        int jobsCanceled = 0;
        for (Future<List<Feature>> job : ftList)
            if (job.cancel(true))
                jobsCanceled++;
        log.error("Canceled {} feature jobs because they took too long",
                        jobsCanceled);
        break;
    }
}

我在这段代码中遇到的问题是,在等待port.getXXXFeatures(…)返回时,实际上并没有取消可调用项,而是以某种方式保持运行。正如你从照片上看到的 if (Thread.currentThread().isInterrupted()) log.error("XXX was interrupted"); 使用中断标志的语句 是 在port.getFeatures返回后设置,这仅在Webservice调用正常完成后可用,而不是在调用Cancel时中断。

有人能告诉我我做错了什么,以及如何在给定的时间段后停止正在运行的CXF Webservice调用,并在我的应用程序中注册此信息吗?

向你问好,蒂姆