gaspersic/java-design-patterns
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

Reformat Async Method Invocation - Issue #224

Browse Source
This commit is contained in:
Ankur Kaushal 2015-11-01 17:33:25 -05:00
parent 95c16200e7
commit 16a8c85af6
6 changed files with 229 additions and 228 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

139
async-method-invocation/src/main/java/com/iluwatar/async/method/invocation/App.java
View File

@ -4,23 +4,24 @@ import java.util.concurrent.Callable;
/**
* This application demonstrates the async method invocation pattern. Key parts of the pattern are
* <code>AsyncResult</code> which is an intermediate container for an asynchronously evaluated value,
* <code>AsyncCallback</code> which can be provided to be executed on task completion and
* <code>AsyncResult</code> which is an intermediate container for an asynchronously evaluated
* value, <code>AsyncCallback</code> which can be provided to be executed on task completion and
* <code>AsyncExecutor</code> that manages the execution of the async tasks.
* <p>
* The main method shows example flow of async invocations. The main thread starts multiple tasks with
* variable durations and then continues its own work. When the main thread has done it's job it collects
* the results of the async tasks. Two of the tasks are handled with callbacks, meaning the callbacks are
* executed immediately when the tasks complete.
* The main method shows example flow of async invocations. The main thread starts multiple tasks
* with variable durations and then continues its own work. When the main thread has done it's job
* it collects the results of the async tasks. Two of the tasks are handled with callbacks, meaning
* the callbacks are executed immediately when the tasks complete.
* <p>
* Noteworthy difference of thread usage between the async results and callbacks is that the async results
* are collected in the main thread but the callbacks are executed within the worker threads. This should be
* noted when working with thread pools.
* Noteworthy difference of thread usage between the async results and callbacks is that the async
* results are collected in the main thread but the callbacks are executed within the worker
* threads. This should be noted when working with thread pools.
* <p>
* Java provides its own implementations of async method invocation pattern. FutureTask, CompletableFuture
* and ExecutorService are the real world implementations of this pattern. But due to the nature of parallel
* programming, the implementations are not trivial. This example does not take all possible scenarios into
* account but rather provides a simple version that helps to understand the pattern.
* Java provides its own implementations of async method invocation pattern. FutureTask,
* CompletableFuture and ExecutorService are the real world implementations of this pattern. But due
* to the nature of parallel programming, the implementations are not trivial. This example does not
* take all possible scenarios into account but rather provides a simple version that helps to
* understand the pattern.
*
* @see AsyncResult
* @see AsyncCallback
@ -32,66 +33,68 @@ import java.util.concurrent.Callable;
*/
public class App {
public static void main(String[] args) throws Exception {
// construct a new executor that will run async tasks
AsyncExecutor executor = new ThreadAsyncExecutor();
public static void main(String[] args) throws Exception {
// construct a new executor that will run async tasks
AsyncExecutor executor = new ThreadAsyncExecutor();
// start few async tasks with varying processing times, two last with callback handlers
AsyncResult<Integer> asyncResult1 = executor.startProcess(lazyval(10, 500));
AsyncResult<String> asyncResult2 = executor.startProcess(lazyval("test", 300));
AsyncResult<Long> asyncResult3 = executor.startProcess(lazyval(50L, 700));
AsyncResult<Integer> asyncResult4 = executor.startProcess(lazyval(20, 400), callback("Callback result 4"));
AsyncResult<String> asyncResult5 = executor.startProcess(lazyval("callback", 600), callback("Callback result 5"));
// start few async tasks with varying processing times, two last with callback handlers
AsyncResult<Integer> asyncResult1 = executor.startProcess(lazyval(10, 500));
AsyncResult<String> asyncResult2 = executor.startProcess(lazyval("test", 300));
AsyncResult<Long> asyncResult3 = executor.startProcess(lazyval(50L, 700));
AsyncResult<Integer> asyncResult4 =
executor.startProcess(lazyval(20, 400), callback("Callback result 4"));
AsyncResult<String> asyncResult5 =
executor.startProcess(lazyval("callback", 600), callback("Callback result 5"));
// emulate processing in the current thread while async tasks are running in their own threads
Thread.sleep(350); // Oh boy I'm working hard here
log("Some hard work done");
// emulate processing in the current thread while async tasks are running in their own threads
Thread.sleep(350); // Oh boy I'm working hard here
log("Some hard work done");
// wait for completion of the tasks
Integer result1 = executor.endProcess(asyncResult1);
String result2 = executor.endProcess(asyncResult2);
Long result3 = executor.endProcess(asyncResult3);
asyncResult4.await();
asyncResult5.await();
// wait for completion of the tasks
Integer result1 = executor.endProcess(asyncResult1);
String result2 = executor.endProcess(asyncResult2);
Long result3 = executor.endProcess(asyncResult3);
asyncResult4.await();
asyncResult5.await();
// log the results of the tasks, callbacks log immediately when complete
log("Result 1: " + result1);
log("Result 2: " + result2);
log("Result 3: " + result3);
}
// log the results of the tasks, callbacks log immediately when complete
log("Result 1: " + result1);
log("Result 2: " + result2);
log("Result 3: " + result3);
}
/**
* Creates a callable that lazily evaluates to given value with artificial delay.
*
* @param value value to evaluate
* @param delayMillis artificial delay in milliseconds
* @return new callable for lazy evaluation
*/
private static <T> Callable<T> lazyval(T value, long delayMillis) {
return () -> {
Thread.sleep(delayMillis);
log("Task completed with: " + value);
return value;
};
}
/**
* Creates a callable that lazily evaluates to given value with artificial delay.
*
* @param value value to evaluate
* @param delayMillis artificial delay in milliseconds
* @return new callable for lazy evaluation
*/
private static <T> Callable<T> lazyval(T value, long delayMillis) {
return () -> {
Thread.sleep(delayMillis);
log("Task completed with: " + value);
return value;
};
}
/**
* Creates a simple callback that logs the complete status of the async result.
*
* @param name callback name
* @return new async callback
*/
private static <T> AsyncCallback<T> callback(String name) {
return (value, ex) -> {
if (ex.isPresent()) {
log(name + " failed: " + ex.map(Exception::getMessage).orElse(""));
} else {
log(name + ": " + value);
}
};
}
/**
* Creates a simple callback that logs the complete status of the async result.
*
* @param name callback name
* @return new async callback
*/
private static <T> AsyncCallback<T> callback(String name) {
return (value, ex) -> {
if (ex.isPresent()) {
log(name + " failed: " + ex.map(Exception::getMessage).orElse(""));
} else {
log(name + ": " + value);
}
};
}
private static void log(String msg) {
System.out.println(String.format("[%1$-10s] - %2$s", Thread.currentThread().getName(), msg));
}
private static void log(String msg) {
System.out.println(String.format("[%1$-10s] - %2$s", Thread.currentThread().getName(), msg));
}
}

15
async-method-invocation/src/main/java/com/iluwatar/async/method/invocation/AsyncCallback.java
View File

@ -11,12 +11,11 @@ import java.util.Optional;
*/
public interface AsyncCallback<T> {
/**
* Complete handler which is executed when async task is completed or fails execution.
*
* @param value the evaluated value from async task, undefined when execution fails
* @param ex empty value if execution succeeds, some exception if executions fails
*/
void onComplete(T value, Optional<Exception> ex);
/**
* Complete handler which is executed when async task is completed or fails execution.
*
* @param value the evaluated value from async task, undefined when execution fails
* @param ex empty value if execution succeeds, some exception if executions fails
*/
void onComplete(T value, Optional<Exception> ex);
}

53
async-method-invocation/src/main/java/com/iluwatar/async/method/invocation/AsyncExecutor.java
View File

@ -10,33 +10,32 @@ import java.util.concurrent.ExecutionException;
*/
public interface AsyncExecutor {
/**
* Starts processing of an async task. Returns immediately with async result.
*
* @param task task to be executed asynchronously
* @return async result for the task
*/
<T> AsyncResult<T> startProcess(Callable<T> task);
/**
* Starts processing of an async task. Returns immediately with async result.
*
* @param task task to be executed asynchronously
* @return async result for the task
*/
<T> AsyncResult<T> startProcess(Callable<T> task);
/**
* Starts processing of an async task. Returns immediately with async result. Executes callback
* when the task is completed.
*
* @param task task to be executed asynchronously
* @param callback callback to be executed on task completion
* @return async result for the task
*/
<T> AsyncResult<T> startProcess(Callable<T> task, AsyncCallback<T> callback);
/**
* Ends processing of an async task. Blocks the current thread if necessary and returns the
* evaluated value of the completed task.
*
* @param asyncResult async result of a task
* @return evaluated value of the completed task
* @throws ExecutionException if execution has failed, containing the root cause
* @throws InterruptedException if the execution is interrupted
*/
<T> T endProcess(AsyncResult<T> asyncResult) throws ExecutionException, InterruptedException;
/**
* Starts processing of an async task. Returns immediately with async result. Executes callback
* when the task is completed.
*
* @param task task to be executed asynchronously
* @param callback callback to be executed on task completion
* @return async result for the task
*/
<T> AsyncResult<T> startProcess(Callable<T> task, AsyncCallback<T> callback);
/**
* Ends processing of an async task. Blocks the current thread if necessary and returns the
* evaluated value of the completed task.
*
* @param asyncResult async result of a task
* @return evaluated value of the completed task
* @throws ExecutionException if execution has failed, containing the root cause
* @throws InterruptedException if the execution is interrupted
*/
<T> T endProcess(AsyncResult<T> asyncResult) throws ExecutionException, InterruptedException;
}

40
async-method-invocation/src/main/java/com/iluwatar/async/method/invocation/AsyncResult.java
View File

@ -10,26 +10,26 @@ import java.util.concurrent.ExecutionException;
*/
public interface AsyncResult<T> {
/**
* Status of the async task execution.
*
* @return <code>true</code> if execution is completed or failed
*/
boolean isCompleted();
/**
* Status of the async task execution.
*
* @return <code>true</code> if execution is completed or failed
*/
boolean isCompleted();
/**
* Gets the value of completed async task.
*
* @return evaluated value or throws ExecutionException if execution has failed
* @throws ExecutionException if execution has failed, containing the root cause
* @throws IllegalStateException if execution is not completed
*/
T getValue() throws ExecutionException;
/**
* Gets the value of completed async task.
*
* @return evaluated value or throws ExecutionException if execution has failed
* @throws ExecutionException if execution has failed, containing the root cause
* @throws IllegalStateException if execution is not completed
*/
T getValue() throws ExecutionException;
/**
* Blocks the current thread until the async task is completed.
*
* @throws InterruptedException if the execution is interrupted
*/
void await() throws InterruptedException;
/**
* Blocks the current thread until the async task is completed.
*
* @throws InterruptedException if the execution is interrupted
*/
void await() throws InterruptedException;
}

199
async-method-invocation/src/main/java/com/iluwatar/async/method/invocation/ThreadAsyncExecutor.java
View File

@ -12,116 +12,117 @@ import java.util.concurrent.atomic.AtomicInteger;
*/
public class ThreadAsyncExecutor implements AsyncExecutor {
/** Index for thread naming */
private final AtomicInteger idx = new AtomicInteger(0);
/** Index for thread naming */
private final AtomicInteger idx = new AtomicInteger(0);
@Override
public <T> AsyncResult<T> startProcess(Callable<T> task) {
return startProcess(task, null);
}
@Override
public <T> AsyncResult<T> startProcess(Callable<T> task) {
return startProcess(task, null);
}
@Override
public <T> AsyncResult<T> startProcess(Callable<T> task, AsyncCallback<T> callback) {
CompletableResult<T> result = new CompletableResult<>(callback);
new Thread(() -> {
try {
result.setValue(task.call());
} catch (Exception ex) {
result.setException(ex);
}
}, "executor-" + idx.incrementAndGet()).start();
return result;
}
@Override
public <T> AsyncResult<T> startProcess(Callable<T> task, AsyncCallback<T> callback) {
CompletableResult<T> result = new CompletableResult<>(callback);
new Thread(() -> {
try {
result.setValue(task.call());
} catch (Exception ex) {
result.setException(ex);
}
}, "executor-" + idx.incrementAndGet()).start();
return result;
}
@Override
public <T> T endProcess(AsyncResult<T> asyncResult) throws ExecutionException, InterruptedException {
if (asyncResult.isCompleted()) {
return asyncResult.getValue();
} else {
asyncResult.await();
return asyncResult.getValue();
}
}
@Override
public <T> T endProcess(AsyncResult<T> asyncResult) throws ExecutionException,
InterruptedException {
if (asyncResult.isCompleted()) {
return asyncResult.getValue();
} else {
asyncResult.await();
return asyncResult.getValue();
}
}
/**
* Simple implementation of async result that allows completing it successfully with a value
* or exceptionally with an exception. A really simplified version from its real life cousins
* FutureTask and CompletableFuture.
*
* @see java.util.concurrent.FutureTask
* @see java.util.concurrent.CompletableFuture
*/
private static class CompletableResult<T> implements AsyncResult<T> {
/**
* Simple implementation of async result that allows completing it successfully with a value or
* exceptionally with an exception. A really simplified version from its real life cousins
* FutureTask and CompletableFuture.
*
* @see java.util.concurrent.FutureTask
* @see java.util.concurrent.CompletableFuture
*/
private static class CompletableResult<T> implements AsyncResult<T> {
static final int RUNNING = 1;
static final int FAILED = 2;
static final int COMPLETED = 3;
static final int RUNNING = 1;
static final int FAILED = 2;
static final int COMPLETED = 3;
final Object lock;
final Optional<AsyncCallback<T>> callback;
final Object lock;
final Optional<AsyncCallback<T>> callback;
volatile int state = RUNNING;
T value;
Exception exception;
volatile int state = RUNNING;
T value;
Exception exception;
CompletableResult(AsyncCallback<T> callback) {
this.lock = new Object();
this.callback = Optional.ofNullable(callback);
}
CompletableResult(AsyncCallback<T> callback) {
this.lock = new Object();
this.callback = Optional.ofNullable(callback);
}
/**
* Sets the value from successful execution and executes callback if available. Notifies
* any thread waiting for completion.
*
* @param value value of the evaluated task
*/
void setValue(T value) {
this.value = value;
this.state = COMPLETED;
this.callback.ifPresent(ac -> ac.onComplete(value, Optional.<Exception>empty()));
synchronized (lock) {
lock.notifyAll();
}
}
/**
* Sets the value from successful execution and executes callback if available. Notifies any
* thread waiting for completion.
*
* @param value value of the evaluated task
*/
void setValue(T value) {
this.value = value;
this.state = COMPLETED;
this.callback.ifPresent(ac -> ac.onComplete(value, Optional.<Exception>empty()));
synchronized (lock) {
lock.notifyAll();
}
}
/**
* Sets the exception from failed execution and executes callback if available. Notifies
* any thread waiting for completion.
*
* @param exception exception of the failed task
*/
void setException(Exception exception) {
this.exception = exception;
this.state = FAILED;
this.callback.ifPresent(ac -> ac.onComplete(null, Optional.of(exception)));
synchronized (lock) {
lock.notifyAll();
}
}
/**
* Sets the exception from failed execution and executes callback if available. Notifies any
* thread waiting for completion.
*
* @param exception exception of the failed task
*/
void setException(Exception exception) {
this.exception = exception;
this.state = FAILED;
this.callback.ifPresent(ac -> ac.onComplete(null, Optional.of(exception)));
synchronized (lock) {
lock.notifyAll();
}
}
@Override
public boolean isCompleted() {
return (state > RUNNING);
}
@Override
public boolean isCompleted() {
return (state > RUNNING);
}
@Override
public T getValue() throws ExecutionException {
if (state == COMPLETED) {
return value;
} else if (state == FAILED) {
throw new ExecutionException(exception);
} else {
throw new IllegalStateException("Execution not completed yet");
}
}
@Override
public T getValue() throws ExecutionException {
if (state == COMPLETED) {
return value;
} else if (state == FAILED) {
throw new ExecutionException(exception);
} else {
throw new IllegalStateException("Execution not completed yet");
}
}
@Override
public void await() throws InterruptedException {
synchronized (lock) {
if (!isCompleted()) {
lock.wait();
}
}
}
}
@Override
public void await() throws InterruptedException {
synchronized (lock) {
if (!isCompleted()) {
lock.wait();
}
}
}
}
}

11
async-method-invocation/src/test/java/com/iluwatar/async/method/invocation/AppTest.java
View File

@ -9,10 +9,9 @@ import org.junit.Test;
*/
public class AppTest {
@Test
public void test() throws Exception {
String[] args = {};
App.main(args);
}
@Test
public void test() throws Exception {
String[] args = {};
App.main(args);
}
}