在Java Future上使用带有isDone和Cancel的轮询，而不是阻塞get

Question

我正在处理一些我不能立即重构的遗留代码。

代码使用阻塞Java未来。它使用future.get（with TimeOut，...）。这意味着我们需要有一个合适大小的线程池来进行足够的响应。因为调用将被阻塞，直到它们完成或超时。

问：我在考虑获取未来并将其放入一个数据结构中，该结构将知道任务的执行开始。然后有一个专用线程或池，它将在数据结构上循环并检查是否future.isDone或已超过超时限制。如果是，它可以获取结果或取消执行。这样就不需要太多线程了。这是一个正确的实现还是根本不推荐？

提前致谢。

编辑：

只是为了提供更多的背景。这些线程用于记录下游服务。我们真的不关心反应，但我们不想联系到洪。因此，我们需要抓住未来，确保它被取消或超时。

这是我在提出问题后写的一个基本模拟。

@Component
public class PollingService {

    private ExecutorService executorService = Executors.newFixedThreadPool(1);
    PoorMultiplexer poorMultiplexer = new PoorMultiplexer();
    private ConcurrentMap<Integer, Map<Future, Long>> futures = new ConcurrentHashMap<>();

    public void startHandler(){
        Thread handler = new Thread(new Runnable() {
            @Override
            public void run() {
                while(true){
                    try {
                        //This should be handled better. If there is not anything stop and re-start it later.
                        Thread.sleep(200);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                    for(Iterator<ConcurrentMap.Entry<Integer, Map<Future, Long>>> it = futures.entrySet().iterator(); it.hasNext();){
                        ConcurrentMap.Entry<Integer, Map<Future, Long>> entry = it.next();
                        Map<Future, Long> futureMap = entry.getValue();
                        boolean isProcessed = false;
                        if(futureMap.keySet().iterator().next().isDone()){
                            //mark completed
                            isProcessed = true;
                        }

                        if(futureMap.values().iterator().next() < (300 + System.currentTimeMillis()) && !isProcessed){
                            //cancel
                            futureMap.keySet().iterator().next().cancel(true);
                            isProcessed = true;
                        }

                        if(isProcessed){
                            futures.remove(entry.getKey());
                            System.out.println("Completed : " + entry.getKey());
                        }

                    }

                    System.out.println("Run completed");
                }
            }
        });

        handler.start();
    }

    public void run(int i) throws InterruptedException, ExecutionException{
        System.out.println("Starting : " + i);

        poorMultiplexer.send(new Runnable() {
            @Override
            public void run() {
                long startTime = System.currentTimeMillis();
                Future future = poorMultiplexer.send(execute());

                Map<Future, Long> entry = new HashMap<>();
                entry.put(future, startTime);
                futures.put(i, entry);
                System.out.println("Added : " + i);
            }
        });
    }

    public void stop(){
        executorService.shutdown();
    }

    public Runnable execute(){
        Worker worker = new Worker();
        return worker;
    }
}


//This is a placeholder for a framework
class PoorMultiplexer {
    private ExecutorService executorService = Executors.newFixedThreadPool(20);

    public Future send(Runnable task){
        return executorService.submit(task);
    }
}


class Worker implements Runnable{

    @Override
    public void run() {
        //service call here
    }

}

Answer 1

在我看来，使用单独的线程异步轮询一组未来确实是一个合理的实现。也就是说，如果您能够添加库依赖项，您可能会发现切换到Guava的ListenableFuture更容易，因为Guava为异步工作提供了大量实用工具。