假设我有一个充满任务的队列，我需要将这些任务提交给执行器服务。我希望他们一次处理一个。我能想到的最简单的方法是：
但是，我试图完全避免阻塞。如果我有 10,000 个这样的队列，它们需要一次处理一个任务，我将用完堆栈空间，因为它们中的大多数将保留阻塞的线程。
我想要的是提交一个任务并提供一个在任务完成时调用的回调。我将使用该回调通知作为发送下一个任务的标志。（functionaljava 和 jetlang 显然使用了这样的非阻塞算法，但我看不懂他们的代码）
如果不编写自己的执行程序服务，我该如何使用 JDK 的 java.util.concurrent 来做到这一点？
（为我提供这些任务的队列本身可能会阻塞，但这是稍后要解决的问题）
定义一个回调接口来接收你想在完成通知中传递的任何参数。然后在任务结束时调用它。
您甚至可以为 Runnable 任务编写一个通用包装器，并将它们提交到ExecutorService. 或者，请参阅下面的 Java 8 内置机制。
class CallbackTask implements Runnable {
private final Runnable task;
private final Callback callback;
CallbackTask(Runnable task, Callback callback) {
this.task = task;
this.callback = callback;
}
public void run() {
task.run();
callback.complete();
}
}
有了CompletableFuture，Java 8 包含了一种更精细的方法来组合管道，其中流程可以异步和有条件地完成。这是一个人为但完整的通知示例。
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ThreadLocalRandom;
import java.util.concurrent.TimeUnit;
public class GetTaskNotificationWithoutBlocking {
public static void main(String... argv) throws Exception {
ExampleService svc = new ExampleService();
GetTaskNotificationWithoutBlocking listener = new GetTaskNotificationWithoutBlocking();
CompletableFuture<String> f = CompletableFuture.supplyAsync(svc::work);
f.thenAccept(listener::notify);
System.out.println("Exiting main()");
}
void notify(String msg) {
System.out.println("Received message: " + msg);
}
}
class ExampleService {
String work() {
sleep(7000, TimeUnit.MILLISECONDS); /* Pretend to be busy... */
char[] str = new char[5];
ThreadLocalRandom current = ThreadLocalRandom.current();
for (int idx = 0; idx < str.length; ++idx)
str[idx] = (char) ('A' + current.nextInt(26));
String msg = new String(str);
System.out.println("Generated message: " + msg);
return msg;
}
public static void sleep(long average, TimeUnit unit) {
String name = Thread.currentThread().getName();
long timeout = Math.min(exponential(average), Math.multiplyExact(10, average));
System.out.printf("%s sleeping %d %s...%n", name, timeout, unit);
try {
unit.sleep(timeout);
System.out.println(name + " awoke.");
} catch (InterruptedException abort) {
Thread.currentThread().interrupt();
System.out.println(name + " interrupted.");
}
}
public static long exponential(long avg) {
return (long) (avg * -Math.log(1 - ThreadLocalRandom.current().nextDouble()));
}
}
在 Java 8 中，您可以使用CompletableFuture。这是我的代码中的一个示例，我使用它从我的用户服务中获取用户，将它们映射到我的视图对象，然后更新我的视图或显示错误对话框（这是一个 GUI 应用程序）：
CompletableFuture.supplyAsync(
userService::listUsers
).thenApply(
this::mapUsersToUserViews
).thenAccept(
this::updateView
).exceptionally(
throwable -> { showErrorDialogFor(throwable); return null; }
);
它异步执行。我正在使用两种私有方法：mapUsersToUserViews和updateView.
使用Guava 的可听未来 API并添加回调。参照。从网站：
ListeningExecutorService service = MoreExecutors.listeningDecorator(Executors.newFixedThreadPool(10));
ListenableFuture<Explosion> explosion = service.submit(new Callable<Explosion>() {
public Explosion call() {
return pushBigRedButton();
}
});
Futures.addCallback(explosion, new FutureCallback<Explosion>() {
// we want this handler to run immediately after we push the big red button!
public void onSuccess(Explosion explosion) {
walkAwayFrom(explosion);
}
public void onFailure(Throwable thrown) {
battleArchNemesis(); // escaped the explosion!
}
});
您可以扩展FutureTask类，并覆盖该done()方法，然后将FutureTask对象添加到ExecutorService，因此该done()方法将在FutureTask立即完成时调用。
ThreadPoolExecutor还有可以覆盖和使用的钩子方法beforeExecute。afterExecute这是来自JavadocsThreadPoolExecutor的描述。
挂钩方法
此类提供受保护的可重写
beforeExecute(java.lang.Thread, java.lang.Runnable)和afterExecute(java.lang.Runnable, java.lang.Throwable)在执行每个任务之前和之后调用的方法。这些可用于操纵执行环境；例如，重新初始化ThreadLocals、收集统计信息或添加日志条目。此外，方法terminated()可以被覆盖以执行任何需要在Executor完全终止后完成的特殊处理。如果钩子或回调方法抛出异常，内部工作线程可能会依次失败并突然终止。
它来自java.util.concurrent并且正是在继续之前等待多个线程完成执行的方式。
为了实现您所关注的回调效果，这确实需要一些额外的工作。也就是说，您自己在一个单独的线程中处理这个，该线程使用CountDownLatch并等待它，然后继续通知您需要通知的任何内容。没有对回调或任何类似效果的本地支持。
编辑：既然我进一步理解了您的问题，我认为您不必要地走得太远了。如果你接受一个常规的SingleThreadExecutor，给它所有的任务，它会在本地进行排队。
如果您想确保不会同时运行任何任务，请使用SingleThreadedExecutor。任务将按照提交的顺序进行处理。您甚至不需要保留任务，只需将它们提交给执行官即可。
这是使用 Guava 对 Pache 答案的扩展ListenableFuture。
特别是，Futures.transform()return ListenableFutureso 可用于链接异步调用。Futures.addCallback()return void，因此不能用于链接，但有利于处理异步完成时的成功/失败。
// ListenableFuture1: Open Database
ListenableFuture<Database> database = service.submit(() -> openDatabase());
// ListenableFuture2: Query Database for Cursor rows
ListenableFuture<Cursor> cursor =
Futures.transform(database, database -> database.query(table, ...));
// ListenableFuture3: Convert Cursor rows to List<Foo>
ListenableFuture<List<Foo>> fooList =
Futures.transform(cursor, cursor -> cursorToFooList(cursor));
// Final Callback: Handle the success/errors when final future completes
Futures.addCallback(fooList, new FutureCallback<List<Foo>>() {
public void onSuccess(List<Foo> foos) {
doSomethingWith(foos);
}
public void onFailure(Throwable thrown) {
log.error(thrown);
}
});
注意：除了链接异步任务之外，Futures.transform()还允许您将每个任务安排在单独的执行程序上（此示例中未显示）。
实现Callback机制的简单代码ExecutorService
import java.util.concurrent.*;
import java.util.*;
public class CallBackDemo{
public CallBackDemo(){
System.out.println("creating service");
ExecutorService service = Executors.newFixedThreadPool(5);
try{
for ( int i=0; i<5; i++){
Callback callback = new Callback(i+1);
MyCallable myCallable = new MyCallable((long)i+1,callback);
Future<Long> future = service.submit(myCallable);
//System.out.println("future status:"+future.get()+":"+future.isDone());
}
}catch(Exception err){
err.printStackTrace();
}
service.shutdown();
}
public static void main(String args[]){
CallBackDemo demo = new CallBackDemo();
}
}
class MyCallable implements Callable<Long>{
Long id = 0L;
Callback callback;
public MyCallable(Long val,Callback obj){
this.id = val;
this.callback = obj;
}
public Long call(){
//Add your business logic
System.out.println("Callable:"+id+":"+Thread.currentThread().getName());
callback.callbackMethod();
return id;
}
}
class Callback {
private int i;
public Callback(int i){
this.i = i;
}
public void callbackMethod(){
System.out.println("Call back:"+i);
// Add your business logic
}
}
输出：
creating service
Callable:1:pool-1-thread-1
Call back:1
Callable:3:pool-1-thread-3
Callable:2:pool-1-thread-2
Call back:2
Callable:5:pool-1-thread-5
Call back:5
Call back:3
Callable:4:pool-1-thread-4
Call back:4
关键说明：
newFixedThreadPool(5) 为newFixedThreadPool(1) 如果您想在分析上一个任务的结果后处理下callback一个任务，只需在下面的行中取消注释
//System.out.println("future status:"+future.get()+":"+future.isDone());
您可以替换newFixedThreadPool()为其中之一
Executors.newCachedThreadPool()
Executors.newWorkStealingPool()
ThreadPoolExecutor
取决于您的用例。
如果要异步处理回调方法
一个。将共享传递ExecutorService or ThreadPoolExecutor给 Callable 任务
湾。将您的Callable方法转换为Callable/Runnable任务
C。推送回调任务到 ExecutorService or ThreadPoolExecutor
只是添加到马特的回答，这有帮助，这里有一个更充实的例子来展示回调的使用。
private static Primes primes = new Primes();
public static void main(String[] args) throws InterruptedException {
getPrimeAsync((p) ->
System.out.println("onPrimeListener; p=" + p));
System.out.println("Adios mi amigito");
}
public interface OnPrimeListener {
void onPrime(int prime);
}
public static void getPrimeAsync(OnPrimeListener listener) {
CompletableFuture.supplyAsync(primes::getNextPrime)
.thenApply((prime) -> {
System.out.println("getPrimeAsync(); prime=" + prime);
if (listener != null) {
listener.onPrime(prime);
}
return prime;
});
}
输出是：
getPrimeAsync(); prime=241
onPrimeListener; p=241
Adios mi amigito
您可以使用 Callable 的实现，这样
public class MyAsyncCallable<V> implements Callable<V> {
CallbackInterface ci;
public MyAsyncCallable(CallbackInterface ci) {
this.ci = ci;
}
public V call() throws Exception {
System.out.println("Call of MyCallable invoked");
System.out.println("Result = " + this.ci.doSomething(10, 20));
return (V) "Good job";
}
}
CallbackInterface 是非常基本的东西
public interface CallbackInterface {
public int doSomething(int a, int b);
}
现在主类看起来像这样
ExecutorService ex = Executors.newFixedThreadPool(2);
MyAsyncCallable<String> mac = new MyAsyncCallable<String>((a, b) -> a + b);
ex.submit(mac);