Java 中的多线程帮助

Question

我是 Java 新手，我需要一些帮助来完成这个程序。这是一个大型类项目的一小部分，我必须使用多线程。

这是我想要在算法上做的事情：

while (there is still input left, store chunk of input in <chunk>)
{
    if there is not a free thread in my array then
      wait until a thread finishes

    else there is a free thread then
       apply the free thread to <chunk> (which will do something to chunk and output it).  
              Note: The ordering of the chunks being output must be the same as input
}

那么，我不知道该怎么做的主要事情：

• 如何检查数组中是否有空闲线程？我知道有一个函数 ThreadAlive，但每次在循环中轮询每个线程似乎效率非常低。
• 如果没有空闲线程，我如何才能等到一个线程完成？
• 顺序很重要。如何保留线程输出的顺序？如图所示，输出的顺序需要与输入的顺序匹配。如何保证这种同步？
• 我如何将块传递到我的线程？我可以只使用 Runnable 接口来做到这一点吗？

非常感谢任何有关这四个项目符号的帮助。因为我是一个超级菜鸟，所以代码示例会有很大帮助。

（旁注：创建一个线程数组只是我的一个想法，用于处理用户定义的线程数。如果您有更好的方法来处理这个问题，欢迎提出建议！）

Answer 1

听起来你基本上有一个生产者/消费者模型，可以通过 ExecutorService 和 BlockingQueue 来解决。这是一个具有类似答案的类似问题：

生产者/消费者工作队列

Answer 2

正如 @altaiojok 提到的，您想要使用 ExecutorService 和 阻塞队列。基本算法的工作原理如下：

ExecutorService executor = Executors.newFixedThreadPool(...); //or newCachedThreadPool, etc...
BlockingQueue<Future<?>> outputQueue = new LinkedBlockingQueue<Future<?>>();


//To be run by an input processing thread
void submitTasks() {
    BufferedReader input = ... //going to assume you have a file that you want to read, this could be any method of input (file, keyboard, network, etc...)
    String line = input.readLine();
    while(line != null) {
        outputQueue.add(executor.submit(new YourCallableImplementation(line)));
        line = input.readLine();
    }
}

//To be run by a different output processing thread
void processTaskOutput() {
    try {
        while(true) {
            Future<?> resultFuture = outputQueue.take();
            ? result = resultFuture.get();
            //process the output (write to file, send to network, print to screen, etc...
        }
    } catch (InterruptedException e) {
        Thread.currentThread().interrupt();
    }
}

我将让您弄清楚如何实现 Runnable 来创建输入和输出线程以及如何实现 Callable您需要处理的任务。

Answer 3

我建议使用 commons-pool 它提供线程池，以便您可以轻松限制使用的线程数量它还提供了一些其他辅助方法。

关于排序：查看 synchronize 关键字。

我建议看一下java教程（关于并发的部分）： http://download.oracle.com/javase/tutorial/essential/concurrency/index.html

Answer 4

流可能会派上用场：

List<Chunk> chunks = new ArrayList<>();
//....
Function<Chunk, String> toWeightInfo = (chunk) -> "weight = "+(chunk.size()*chunk.prio());

List<String> results = chunks.parallelStream()
  .map(toWeightInfo)
  .collect(Collectors.toList());

System.out.println(results);

并行流使用系统默认的“fork/join”线程池，该线程池应该是可用逻辑 CPU 的大小并并行处理您的内容。它还保证结果的顺序相同。
并行流 API 隐藏了将空闲线程分配给作业和优化（例如窃取工作）的所有复杂性。只要给它一些咀嚼的东西，它就会发挥它的魔力。

如果需要使用自定义大小的线程池，请参考
Java 8 并行流中的自定义线程池问题。
您也可以看看这个很好的 Java 8 Stream 教程。

如果您的情况相当复杂，并且您正在将块流式传输到程序中，并且您有多个工作阶段，其中有些必须是串行的，有些可以并行，有些相互依赖，那么您可能会看看< a href="https://lmax-exchange.github.io/disruptor/" rel="nofollow noreferrer">LMAX 的 Disruptor 框架。

亲切的问候

Answer 5

使用 ExecutorCompletionService 和 Future。它们共同提供了一个基于线程池的任务框架，可以解决您的所有问题。

如何检查数组中是否有空闲线程？我知道有一个函数 ThreadAlive，但每次在我的循环中轮询每个线程似乎效率非常低。

你不必这样做。执行器将以（超）高效的方式为您完成此操作。您只需向它提交任务并坐下来即可。

如果没有空闲线程，我如何才能等到一个线程结束？

再说一次，你真的不必这样做。这由执行者负责。

顺序很重要。如何保留线程输出的顺序？如图所示，输出的顺序需要与输入的顺序匹配。我如何保证这种同步？

这是一个问题。如果您希望处理后的输出（用您的话来说是块）以与这些块在初始数组中存在的顺序相同的顺序到达，则必须解决以下几点：

是否只是结果的到达顺序重要或者是任务处理本身依赖于顺序？如果是前者，那么很容易做到，但如果是后者，那就有问题了。 （考虑到您承认自己是 Java 新手，我认为这是非常困难的事情，因此我建议您在尝试此操作之前进行更多学习。）

假设是前一种情况，您可以做的是： 提交按某种顺序将块发送给执行器，每次提交都会为您提供任务处理输出的句柄（称为 Future）。将这些句柄存储在有序队列中，当您需要结果时，对这些 Future 调用 get()。请注意，如果订单中间的某些任务需要很长时间才能完成，那么后续任务的结果也会延迟。

我如何将块传递到我的线程？我可以只使用 Runnable 接口来执行此操作吗？

创建一个 Callable 实例，将每个块包装到实例中。这代表您将 submit() 到 ExecutorService 的任务。