java - 如何将带有 ReentrantLock 的 Thread 包装到 CompletableFuture 调用中？

Question

这是我当前的实现，它连续处理不同的文件读取/保存操作:

public void runThread(MyThreadImpl myThreadImpl) {
    synchronized (this) {
        this.myThreadImpl = myThreadImpl;
        notify();
    }
}

synchronized public void run() {
    while (true)
        try {
            wait();
            Global.myReentrantLock.lock();
            try {
                try {
                    myThreadImpl.call();
                } catch (FileException e) {
                    // trace e
                } catch (RuntimeException e) {
                    // trace e
                } catch (Exception e) {
                    // trace e
                }
            } finally {
                Global.myReentrantLock.unlock();
            }
        } catch (InterruptedException e) {
          // trace e
        } catch (Exception e) {
          // trace e
        }
}

我有一个问题，在执行另一个操作之前我没有等待线程结果，并且我遇到了有必要这样做的情况。

由于我使用的是 Java 8，所以我想将其包装在 CompletableFuture 中。我如何使用当前的实现来做到这一点？

Answer 1

您可以执行以下操作:

• 您可以使用队列，而不是将下一个要完成的作业存储为单个引用 ( this.myThreadImpl )，并在锁释放后更新。
• 添加新作业时，会生成新的 CompletableFuture创建后，对其的引用将返回给调用者。
• 一旦工作完成， future 就完成了。

更新您的代码，并假设 queue是 Queue<Pair<CompletableFuture<Void>, MyThreadImpl>> 类型的阻塞队列，你会:

/**
 * @return a Future that will complete once the passed MyThreadImpl has been run.
 */
public CompletableFuture<Void> runThread(MyThreadImpl myThreadImpl) {
    Pair<CompletableFuture<Void>, MyThreadImpl> p = 
              new Pair<>(new CompletableFuture<>(),myThreadImpl);
    queue.add(p);
    return p.left;
}

public void run() {
    while (true) {
        try {

            Pair<CompletableFuture<MyThreadImpl>, MyThreadImpl> p = 
                  queue.take(); // will block until a job is added

            try {
                p.right.call();
                p.left.complete(null); // Future<Void> can only be completed with null. Alternatively, it could be completed with a relevant result.
            } catch (Exception e) {
                p.left.completeExceptionally(e);
            }

         } catch (InterruptedException e) {
            // trace e
         } 
   }
}

_{在这里，Pair只需要成为一对类似的 pojo 即可。 (例如，它可能是 apache commons 的 ImmutablePair 。)}

当需要处理数据时，阻塞队列通常很有用:https://docs.oracle.com/javase/8/docs/api/java/util/concurrent/BlockingQueue.html

另外，你看过ExecutorService吗？ ？您可以使用基于单线程的线程以串行方式执行作业:它是 submit(Callable<> task)方法很像 runThread()上面定义的，因为它返回 Future<Void>这会告诉您任务何时完成。