500个工作线程，什么样的线程池？

Question

我想知道这是否是最好的方法。我有大约 500 个无限期运行的线程，但在完成一个处理周期后 Thread.sleep 一分钟。

   ExecutorService es = Executors.newFixedThreadPool(list.size()+1);
   for (int i = 0; i < list.size(); i++) {
      es.execute(coreAppVector.elementAt(i)); //coreAppVector is a vector of extends thread objects
   }

正在执行的代码非常简单，基本上就是这样，

class aThread extends Thread {
   public void run(){
      while(true){
         Thread.sleep(ONE_MINUTE);
         //Lots of computation every minute
      }
   }
}

我确实需要为每个正在运行的任务分配一个单独的线程，因此更改架构不是一个选择。我尝试使我的 threadPool 大小等于 Runtime.getRuntime().availableProcessors() ，它尝试运行所有 500 个线程，但只让其中 8 个（4x超线程）执行。其他线程不会投降并让其他线程轮流执行。我尝试添加 wait() 和 notification()，但仍然没有成功。如果有人有一个简单的例子或一些提示，我将不胜感激！

嗯，这个设计可以说是有缺陷的。线程实现遗传编程或 GP，一种学习算法。每个线程都会分析高级趋势并做出预测。如果线程完成，学习的内容就会丢失。也就是说，我希望 sleep() 能够让我在一个线程不“学习”时共享一些资源

所以实际要求是

如何安排维护任务 状态并每 2 分钟运行一次，但是 控制一次执行多少个。

Answer 1

如果您的线程没有终止，这是线程内代码的错误，而不是线程池的错误。要获得更详细的帮助，您需要发布正在执行的代码。

另外，为什么每个线程完成后都让其休眠？让它完成不是更好吗？

此外，我认为您通过使线程数等于您希望执行的任务数来滥用线程池。线程池的要点是对使用的资源数量进行限制；这种方法并不比根本不使用线程池好。

最后，您不需要将 Thread 实例传递给 ExecutorService，只需传递 Runnable 实例即可。 ExecutorService 维护自己的线程池，这些线程无限循环，从内部队列中提取工作（工作是您提交的 Runnable ）。

Answer 2

为什么不使用 ScheduledExecutorService 来 安排每个任务每分钟运行一次，而不是让所有这些线程空闲一整分钟？

ScheduledExecutorService workers = 
  Executors.newScheduledThreadPool(Runtime.getRuntime().availableProcessors());
for (Runnable task : list) { 
  workers.scheduleWithFixedDelay(task, 0, 1, TimeUnit.MINUTES);
}

“改变架构不是一个选择”是什么意思？如果您的意思是您根本无法修改您的任务（具体来说，任务必须循环，而不是运行一次，并且调用 Thread.sleep()无法删除），那么“良好的性能也不是一个选择”。

Answer 3

我不确定您的代码在如何使用线程池方面在语义上是否正确。 ExecutionService 在内部创建和管理线程，客户端只需提供一个 Runnable 实例，其 run() 方法将在池线程之一的上下文中执行。您可以查看我的示例。另请注意，每个正在运行的线程占用约 10Mb 的系统内存用于堆栈，并且在 Linux 上，java 到本机线程的映射是 1 对 1。

Answer 4

您应该让它返回并使用 ThreadPoolexecutor 执行每分钟发布到工作队列的工作。

Answer 5

回答你的问题，什么类型的线程池？

我发表了我的评论，但这确实应该解决您的问题。您的计算可能需要 2 秒才能完成。您有许多任务 (500) 希望尽快完成。假设没有 IO 和/或网络流量，您可以实现的最快吞吐量是使用 Runtime.getRuntime().availableProcessors() 线程数。

如果将线程数增加到 500 个，则每个任务都将在自己的线程上执行，但操作系统会每隔一段时间调度一个线程以分配给另一个线程。也就是说，在任何给定点都会进行 125 次上下文切换。每次上下文切换都会增加每个任务运行的时间。

这里的大局是，当处理器数量远远超过时，添加更多线程并不等于更高的吞吐量。

编辑：快速更新。你不需要睡在这里。当您使用 8 个处理器执行 500 个任务时，每个任务将在 2 秒内完成，完成后，它运行的线程将执行下一个任务并完成该任务。

Answer 6

8 个线程是您的系统可以处理的最大值，如果超过 8 个线程，上下文切换就会减慢您的速度。

看看这篇文章 http://www.informit.com/ articles/article.aspx?p=1339471&seqNum=4 它将为您提供如何操作的概述。

Answer 7

这应该做你想要的，但不是你要求的:-)你必须取出 Thread.sleep()

ScheduledRunnable.java

import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

public class ScheduledRunnable
{
    public static void main(final String[] args)
    {
        final int numTasks = 10;
        final ScheduledExecutorService ses = Executors.newScheduledThreadPool(Runtime.getRuntime().availableProcessors());
        for (int i = 0; i < numTasks; i++)
        {
            ses.scheduleAtFixedRate(new MyRunnable(i), 0, 10, TimeUnit.SECONDS);
        }
    }

    private static class MyRunnable implements Runnable
    {
        private int id;
        private int numRuns;

        private MyRunnable(final int id)
        {
            this.id = id;
            this.numRuns = 0;
        }

        @Override
        public void run()
        {
            this.numRuns += 1;
            System.out.format("%d - %d\n", this.id, this.numRuns);
        }
    }
}

这会每 10 秒安排一次 Runnables 来显示行为。
如果您确实需要在处理完成后等待固定的时间，您可能需要尝试使用您需要的 .scheduleXXX 方法。我认为，fixedWait 只会每 N 次运行一次，无论执行时间是多少。

Answer 8

我确实需要为每个正在运行的任务分配一个单独的线程，因此更改架构不是一个选择。

如果这是真的（例如，调用外部阻塞函数），则为它们创建单独的线程并启动它们。您无法创建具有有限数量线程的线程池，因为其中一个线程中的阻塞函数将阻止将任何其他可运行对象放入其中，并且创建每个任务只有一个线程的线程池不会获得太多好处。

我尝试使我的 threadPool 大小等于 Runtime.getRuntime().availableProcessors() ，它尝试运行所有 500 个线程，但只让其中 8 个（4x 超线程）执行。

当您将创建的 Thread 对象传递给线程池时，它只会看到它们实现了 Runnable。因此，它将运行每个 Runnable 直至完成。任何停止 run() 方法返回的循环都将不允许下一个排队任务运行；例如：

public static void main (String...args) {
    ExecutorService executor = Executors.newFixedThreadPool(2);

    for (int i = 0; i < 10; ++i) {
        final int task = i;

        executor.execute(new Runnable () {
        private long lastRunTime = 0;
            @Override
            public void run () {

                for (int iteration = 0; iteration < 4; )
                {
                    if (System.currentTimeMillis() - this.lastRunTime > TIME_OUT)
                    {
                        // do your work here
                        ++iteration;
                        System.out.printf("Task {%d} iteration {%d} thread {%s}.\n", task, iteration, Thread.currentThread());

                        this.lastRunTime = System.currentTimeMillis();
                    }
                    else
                    {
                        Thread.yield(); // otherwise, let other threads run
                    }
                }
            }
        });
    }

    executor.shutdown();
}

打印出：

Task {0} iteration {1} thread {Thread[pool-1-thread-1,5,main]}.
Task {1} iteration {1} thread {Thread[pool-1-thread-2,5,main]}.
Task {0} iteration {2} thread {Thread[pool-1-thread-1,5,main]}.
Task {1} iteration {2} thread {Thread[pool-1-thread-2,5,main]}.
Task {0} iteration {3} thread {Thread[pool-1-thread-1,5,main]}.
Task {1} iteration {3} thread {Thread[pool-1-thread-2,5,main]}.
Task {0} iteration {4} thread {Thread[pool-1-thread-1,5,main]}.
Task {2} iteration {1} thread {Thread[pool-1-thread-1,5,main]}.
Task {1} iteration {4} thread {Thread[pool-1-thread-2,5,main]}.
Task {3} iteration {1} thread {Thread[pool-1-thread-2,5,main]}.
Task {2} iteration {2} thread {Thread[pool-1-thread-1,5,main]}.
Task {3} iteration {2} thread {Thread[pool-1-thread-2,5,main]}.
Task {2} iteration {3} thread {Thread[pool-1-thread-1,5,main]}.
Task {3} iteration {3} thread {Thread[pool-1-thread-2,5,main]}.
Task {2} iteration {4} thread {Thread[pool-1-thread-1,5,main]}.
...

显示第一个（线程池大小）任务在计划下一个任务之前运行完成。

您需要做的是创建运行一段时间的任务，然后让其他任务运行。如何构建这些取决于您想要实现的目标，

• 是希望所有任务同时运行、所有任务等待一分钟，然后再次同时运行，还是任务之间不同步其他
• 是否确实希望每个任务以一分钟的间隔运行
• 是否您的任务可能会阻塞，因此确实需要单独的线程
• 如果任务阻塞时间超过了运行的预期窗口，
• 则预期会出现什么行为 如果任务阻塞时间长于重复率（阻塞时间超过一分钟）

根据这些问题的答案，可以使用 ScheduledExecutorService、信号量或互斥体的某种组合来协调任务。最简单的情况是非阻塞、非同步任务，在这种情况下，直接使用 ScheduledExecutorService 每分钟运行一次可运行对象。

Answer 9

您可以重写您的项目以使用一些基于代理的并发框架，例如 Akka 吗？

Answer 10

通过将线程数量减少到系统实际可以处理的数量，您当然可以发现吞吐量有所提高。您愿意稍微改变一下螺纹的设计吗？它将减轻调度程序的负担，将休眠线程放入队列中，而不是实际拥有数百个休眠线程。

class RepeatingWorker implements Runnable {

private ExecutorService executor;
private Date lastRan;

//constructor takes your executor

@Override
public void run() {

  try {
    if (now > lastRan + ONE_MINUTE) {
      //do job
      lastRan = now;
    } else {
      return;
  } finally {
    executor.submit(this);
  }
}
}

这保留了“作业无限重复，但在执行之间至少等待一分钟”的核心语义，但现在您可以将线程池​​调整为机器可以处理的内容，而那些不工作的线程则在队列中，而不是徘徊在调度程序中作为休眠线程。如果没有人真正做任何事情，就会出现一些等待繁忙的行为，但我从您的帖子中假设应用程序的整个目的是运行这些线程，并且它当前正在影响您的处理器。如果必须为其他事情留出空间，您可能需要对此进行调整:)

Answer 11

你需要一个信号量。

class AThread extends Thread {
   Semaphore sem;
   AThread(Semaphore sem) {
     this.sem = sem;
   }
   public void run(){
      while(true){
         Thread.sleep(ONE_MINUTE);
         sem.acquire();
         try {
           //Lots of computation every minute
         } finally {
           sem.release();
         }
      }
   }
}

实例化 AThreads 时，您需要传递相同的信号量实例：

Semaphore sem = new Semaphore(MAX_AVAILABLE, true);

编辑：谁投票否决了，请解释一下原因？我的解决方案有问题吗？