java如何在未来释放资源

Question

Uscase

假设我们使用CompletableFuture运行执行。runAsync（…）在runnable中，我们使用try with resources块（无论发生什么情况，我们都使用一些应该关闭的资源），当try块中的执行未完成时，我们会取消可完成的未来。。。虽然执行已停止，但应关闭的资源未关闭AutoClosable的close（）未被调用

---

问题:

这是一个java问题，还是有一种方法可以正确地做到这一点？没有像使用期货（支持中断等）这样的黑客解决方法，如果它的预期行为是，当不可中断的可完成未来被取消时，人们应该如何处理类似的情况

---

密码

public class AutoClosableResourceTest {

    public static class SomeService{
        public void connect(){
            System.out.println("connect");
        }

        public Integer disconnect(){
            System.out.println("disconnect");
            return null;
        }
    }

    public static class AutoClosableResource<T> implements AutoCloseable {

        private final T resource;
        private final Runnable closeFunction;

        private AutoClosableResource(T resource, Runnable closeFunction){
            this.resource = resource;
            this.closeFunction = closeFunction;
        }

        public T get(){
            return resource;
        }

        @Override
        public void close() throws Exception {
            closeFunction.run();
        }
    }

    @Test
    public void testTryWithResource() throws InterruptedException {
        SomeService service  = new SomeService();

        CompletableFuture<Void> async = CompletableFuture.runAsync(() -> {
            try (AutoClosableResource<SomeService> resource = new AutoClosableResource<>(service, service::disconnect)) {
                resource.get().connect();
                while (true) {
                    Thread.sleep(1000);
                    System.out.println("working...");
                }
            } catch (Exception e) {
                e.printStackTrace();
            }
        });

        Thread.sleep(2500);
        async.cancel(true);
        Thread.sleep(2500);

    }
}

这将产生

connect
working...
working...
working...
working...

正如您所看到的，它不调用cancel（）并使资源保持打开状态

Answer 1

我在Java8SE中也面临这个问题。对我来说，重要的是不要使用第三方库

cancel(mayInterruptIfRunning) this value has no effect in this implementation because interrupts are not used to control processing.

的想法是使用线程。调用cancel（）时中断（），但仅适用于可运行的

/** Enable and disable the interrupt */
private static class Interruptor {

    volatile boolean interrupted;
    volatile Runnable interrupt;

    /** Enable interrupt support */
    synchronized boolean start() {
        if (interrupted) {
            return false;
        }
        Thread runThread = Thread.currentThread();
        interrupt = () -> {
            if (runThread != Thread.currentThread()) {
                runThread.interrupt();
            }
        };
        return true;
    }

    /** Interrupt Runnable */
    synchronized void interrupt() {
        if (interrupted) {
            return;
        }
        interrupted = true;
        if (interrupt != null) {
            interrupt.run();
            interrupt = null;
        }
    }

    /** Disable interrupt support */
    synchronized void finish() {
        interrupt = null;
    }
}


/** CompletableFuture with interrupt support */
public static CompletableFuture<Void> runAsyncInterrupted(Runnable run) {

    final Interruptor interruptor = new Interruptor();

    Runnable wrap = () -> {
        if (!interruptor.start()) { // allow interruption
            return; // was canceled before the thread started
        }
        try {
            run.run(); // can be interrupted
        } finally {
            interruptor.finish(); // can no longer be interrupted
        }
    };

    CompletableFuture<Void> cfRun = CompletableFuture.runAsync(wrap);

    // here is caught "CompletableFuture.cancel()"
    cfRun.whenComplete((r, t) -> {
        if (t instanceof CancellationException) {
            interruptor.interrupt();
        }
    });

    return cfRun;
}

使用示例

Runnable mySlowIoRun = () -> {
    try {
        InputStream is = openSomeResource(); // open resource
        try {
            // there may be problem (#1) with reading,
            // such as loss of network connection
            int bt = is.read();
            // ..
            // .. some code
        } finally {
            is.close(); // problem (#2): releases any system resources associated with the stream
        }
    } catch (Throwable th) {
        throw new RuntimeException(th);
    }
};

CompletableFuture<Void> cf = runAsyncInterrupted(mySlowIoRun);

try {
    cf.get(5, TimeUnit.SECONDS); // 5 sec timeout
} catch (Throwable th) {
    cf.cancel(true); // cancel with interrupt mySlowIoRun
    throw th;
}

Answer 2

以下代码将卡在无限循环中。调用异步。cancel不会与以下循环通信，因为它希望停止

while (true) {
    Thread.sleep(1000);
    System.out.println("working...");
}

测试用例退出是因为卡在这个循环中的线程不是守护进程线程

将while循环检查替换为以下内容，即在每次迭代中检查isCancelled标志。呼唤未来。cancel（）会将未来标记为已取消，但不会中断通过runAsync启动的线程

while (isCancelled()) {
    Thread.sleep(1000);
   System.out.println("working...");
}

Answer 3

虽然有一个答案被标记为正确，但原因是完全不同的——请参阅CompletableFuture.cancel(mayInterruptIfRunning){a1}的文档，并阅读文章CompletableFuture can't be interrupted以更好地理解这个问题

这个问题在我的Tascalate Concurrent库中得到了解决，对代码的更改应该是： 从…起 CompletableFuture<Void> async = CompletableFuture.runAsync(() -> { ... });

到 Promise<Void> async = CompletableTask.runAsync(() -> { ... }, someExplicitExecutor); ...您将获得预期的行为（executor线程被中断，AutoClosable被关闭，async用CancellationException完成）

你可以在my blog中阅读更多关于图书馆的信息