不同对象之间的 Java 线程处理答案

【问题标题】：Java Threads handling between separate objects不同对象之间的 Java 线程处理
【发布时间】：2013-01-02 07:20:17
【问题描述】：

当我尝试学习 Java 线程时，我通常会在同一个类中遇到 wait() 和 notify() 的代码示例（实际上几乎所有这些示例都是生产者-消费者示例）。在谷歌搜索各种示例后，不幸的是我找不到我需要的案例：

管理器线程最初创建 n 个线程（并启动它们），其中 http get 请求在单个线程中完成。
对于单个工作线程，完成其生命周期大约需要 20-30 秒。
这里我的经理线程必须知道哪些工人已经完成，并用一个新的替换完成线程。

我想过这样的方法（让 n 为 5）：

List<Runnable> runnables = new ArrayList<Runnable>();
for(int i = 0 ; i < 5 ; i++){
    runnables.add(new MyWorker(params));
}
for(Runnable myWorker : runnables){
   myWorker.run();
}

由于wait() 不支持多个对象，我不能从这里继续下去。另一种解决方案是在为每个工作人员调用一些 isFinished 标志的管理器线程上实现繁忙等待。但我不确定这是否是一种好方法（据我所知这是资源浪费）

【问题讨论】：

您可能正在寻找ThreadPoolExecutor。另请参阅实用程序类 Executors，它具有适用于最常见场景的静态工厂方法。
@fge：您应该将其发布为答案（并详细说明）。
我不会费心等待和通知，这可能很难使用。 java.util.concurrent 包中的高级线程工具更易于使用。
@JBNizet 完成，请修正任何错误；）
你想强制每个线程顺序执行吗？在上面的代码中，您实际上并没有启动新线程，您只是调用每个 Thread 对象的 run 方法。没有发生多线程！真的取决于你想做什么。

标签： java multithreading

【解决方案1】：

在 Java 6 及更高版本中有一个即用型工具，以执行器框架的形式。

既然你想拥有一个固定的线程池，你最好的选择是使用：

ExecutorService service = Executors.newFixedThreadPool(5);

然后您可以使用.execute() 方法（由Executor 接口定义）提交您的Runnable 实例。一旦池中的线程可用，它们将被提交到工作队列并出列。

如果您的各个线程工作方法返回一个值（即，它们实现了Callable<Something>），那么您可以使用.submit() 方法（由ExecutorService 定义，也是扩展Executor 的接口），它将返回一个Future<Something>，您将从中.get() 计算值。

终止线程池有多种方式：.shutdown() 是最基本的，它将同步等待仍然活动的线程终止（并阻止提交新作业）。

Javadocs：Executors、ExecutorService、ThreadPoolExecutor。

其他链接：a book you should buy for all things thread-related（但不包括 Java 7 的 ForkJoinPool）。

PS：多么幸运，sample chapter of the book above (PDF) 涵盖了任务执行；）

【讨论】：

这个 Executor 框架看起来是我想要的，现在我正在阅读如何在完成工作人员后分配工作人员。有一些阻塞队列的例子。除了我的工人不返回任何东西，他们只是做文件 i/o
@px5x2：好吧，在这种情况下使用.execute()。至于线程处理，它是自动完成的，你根本不用担心——这是好事；）你当然可以实现你自己的ThreadPoolExecutor，但你必须有非常具体的需求。

【解决方案2】：

您可以使用semaphore 和wait/notify 方法来完成您想要做的事情。方法是：

使用一次允许的最大线程数初始化semaphore。
等到queue 中的任务可用。
获取semaphore。
运行任务并在完成时发布semaphore。

将所有四个步骤都放在永远的while循环中，您就可以使用任务执行器了。这仅用于学习目的，正如@fge 所说，已经存在ThreadPoolExecuter 可以为您做同样的事情并且更加优化。

import java.util.LinkedList;
import java.util.Queue;
import java.util.concurrent.Semaphore;

public class ThreadExecuter extends Thread {

    Queue<Runnable> tasks = new LinkedList<Runnable>();
    Semaphore s;
    Object bin = new Object();

    public ThreadExecuter(int n) {
        s = new Semaphore(n);
    }

    public void addTask(Runnable r) {
        synchronized (bin) {
            tasks.add(r);
            bin.notifyAll();
        }
    }

    public void run() {
        while (true) {
            try {
                final Runnable r;
                synchronized (bin) {
                    while (tasks.isEmpty())
                        bin.wait();
                    r = tasks.poll();
                }

                s.acquire();
                new Thread() {
                    public void run() {
                        r.run();
                        s.release();
                    }
                }.start();
            } catch (InterruptedException e) {
                // TODO Auto-generated catch block
                e.printStackTrace();
            }
        }

    }
}

main方法如下所示：

import java.util.Random;

public class ThreadTest {

    /**
     * @param args
     */
    public static void main(String[] args) {
            /* to make maximum 10 threads each running 1 runnable */
        ThreadExecuter executer = new ThreadExecuter(10);
        executer.start();

        for(int i = 0; i < 5000; i++) {
                    /* add task in executer, this is non blocking */
            executer.addTask(new Runnable() {
                @Override
                public void run() {
                    System.out.println("Task Executed in " 
                                        + Thread.currentThread().getId());
                    try {
                        Thread.sleep(new Random().nextInt(8000));
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            });
        }
    }

}

【讨论】：