CODEWITHSUNDEEP
javajava-8thread-safety
Saran
Saran

Reputation: 3885

How to wrap Thread with ReentrantLock into CompletableFuture call?

This is my current implementation which handles different file read/save operations consecutively:

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
        }
}

I have a problem that I don't wait for thread result before performing another operation, and I've come to a case where that is necessary.

Since I'm using Java 8, I wanted to wrap this in a CompletableFuture. How can I do this with my current implementation?

Upvotes: 0

Views: 285

Answers (1)

Shastick
Shastick

Reputation: 1258

You could do the following:

  • Instead of storing the next job to be done as a single reference (this.myThreadImpl) that is updated once the lock is free, you could use a queue.
  • When a new job is added, a new CompletableFuture is created, and a reference to it is returned to the caller.
  • Once the job is completed, the future is completed.

Updating your code, and assuming queue is a blocking queue of type Queue<Pair<CompletableFuture<Void>, MyThreadImpl>>, you would have:

/**
 * @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
         } 
   }
}

Here, Pair just needs to be a pair-like pojo. (It could be apache commons's ImmutablePair, for example.)

Blocking queues are generally useful when stuff needs to be processed: https://docs.oracle.com/javase/8/docs/api/java/util/concurrent/BlockingQueue.html

Also, have you looked at ExecutorService? You could use one that is based on a single thread to execute jobs in a serial way: it's submit(Callable<> task) method is quite like runThread() defined above, as it returns a Future<Void> that will tell you when the task is done.

Upvotes: 1

Related Questions