在 Java 中实现这种线程/事件行为的最佳方法是什么？

Question

我有一个线程 (Runnable)，它启动了许多其他线程 (Runnables)。当每个子线程完成时，它需要引发一个事件（或类似的东西）并向父线程返回一个通知。我在 Java (ala C#) 中看不到任何事件 - 我曾希望我可以在父对象中订阅子对象的“我完成事件”，但似乎我做不到。你建议我如何做到这一点？

谢谢

Answer 1

Java 在其线程库中有一个CountDownLatch。创建一个CountDownLatch 并使用您要运行的线程数对其进行初始化。当你创建你的线程时，你应该给他们一个闩锁，每个线程都会在它完成时发出信号。您的主线程将阻塞，直到所有工作线程都完成。

使用CountDownLatch，您将实现与线程的无锁通信。

直接来自 Java 的文档：

 class Driver { // ...
   void main() throws InterruptedException {
     CountDownLatch startSignal = new CountDownLatch(1);
     CountDownLatch doneSignal = new CountDownLatch(N);

     for (int i = 0; i < N; ++i) // create and start threads
       new Thread(new Worker(startSignal, doneSignal)).start();

     doSomethingElse();            // don't let run yet
     startSignal.countDown();      // let all threads proceed
     doSomethingElse();
     doneSignal.await();           // wait for all to finish
   }
 }

 class Worker implements Runnable {
   private final CountDownLatch startSignal;
   private final CountDownLatch doneSignal;
   Worker(CountDownLatch startSignal, CountDownLatch doneSignal) {
      this.startSignal = startSignal;
      this.doneSignal = doneSignal;
   }
   public void run() {
      try {
        startSignal.await();
        doWork();
        doneSignal.countDown();
      } catch (InterruptedException ex) {} // return;
   }

   void doWork() { ... }
 }

http://java.sun.com/j2se/1.5.0/docs/api/java/util/concurrent/CountDownLatch.html

Answer 2

你在你的父对象上创建一个接口

public interface EventListener  {
    void trigger(Object event); 
} 

public class Parent implements EventListener { 
    public synchronized void trigger(Object event) { 
        // process events. 
    }
}

public class Child implements Runnable { 
    private final EventListener listener; 

    public Child(EventListener listen) { 
       listener = listen; 
    }  

    public void run () {
      //do stuff
      listener.trigger( results ); 
    } 
}

Answer 3

您可以在Observer pattern 上使用变体。在父级中实现一个回调函数（例如void finished(SomeArgs args)），并使用对其父级的引用来构造每个子级。当孩子完成后，让它调用父母的finished()方法。

确保回调是线程安全的！

Answer 4

这不使用事件，但这只是我确信实现此目的的许多方法之一。快速警告：为此，您需要将 Runnable 转换为 Thread 对象，或修改界面以在 Runnable 上使用某种 isStopped() 方法，无论您的 Runnable 是否仍在运行，该方法都会返回。

您可以让父线程在列表中跟踪其所有子线程。当子线程结束时，将它计算的值放在某个字段中，比如 result，并创建一个名为 getResult() 的方法。

让父线程定期遍历列表并检查线程是否已停止。如果将 Runnable 转换为 Thread 对象，则有一个名为 isAlive() 的方法来判断线程是否已停止。如果有，请调用 getResult() 并执行任何操作。

在父线程中，您可以这样做：

Boolean running = true;
while (running) {
    //iterate through list
    //if stopped, get value and do whatever
    //if all the child threads are stopped, stop this thread and do whatever
    Thread.sleep(1000); //makes this parent thread pause for 1 second before stopping again
}

Answer 5

java.util.concurrent.ThreadPoolExecutor 类可以满足您的需求。它执行多个线程并提供一个在每个Runnable 完成后调用的钩子。基本上你可以创建一个匿名子类并覆盖afterExecute。像这样：

ThreadPoolExecutor executor = new ThreadPoolExecutor(5, 20, 5,
        TimeUnit.SECONDS, new ArrayBlockingQueue<Runnable>(50)) {
    @Override
    protected void afterExecute(Runnable r, Throwable t) {
        // do your callback stuff here
    }
};

这是完整的例子：

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

public class Main {
    private static int ready = 0;

    public static void main(String[] args) throws InterruptedException {
        ThreadPoolExecutor executor = new ThreadPoolExecutor(5, 20, 5,
                TimeUnit.SECONDS, new ArrayBlockingQueue<Runnable>(50)) {
            @Override
            protected void afterExecute(Runnable r, Throwable t) {
                ready++;
            }
        };

        for (int n = 0; n < 5; n++)
            executor.execute(createTask());
        executor.shutdown();

        while(ready < 5) {
            System.out.println("Ready: " + ready);
            Thread.sleep(100);
        }

        System.out.println("Ready with all.");
    }

    private static Runnable createTask() {
        return new Runnable() {
            @Override
            public void run() {
                try {
                    Thread.sleep((long) (Math.random() * 1000));
                } catch (InterruptedException e) {
                    // ignore exception to make debugging a little harder
                }
            }
        };
    }

}

输出是：

Ready: 0
Ready: 1
Ready: 1
Ready: 3
Ready: 3
Ready: 4
Ready: 4
Ready: 4
Ready: 4
Ready with all.