我最近刚刚将 ArrayBlockingQueue 用于我的多线程进程。但它似乎放慢了速度而不是加速了。你们能帮帮我吗？我基本上是导入一个文件(大约 30 万行)并解析它们并将它们存储在数据库中

public class CellPool {
private static class RejectedHandler implements RejectedExecutionHandler {
    @Override
    public void rejectedExecution(Runnable arg0, ThreadPoolExecutor arg1) {
      System.err.println(Thread.currentThread().getName() + " execution rejected: " + arg0);     
    }
  }

  private static class Task implements Runnable {
    private JSONObject obj;

    public Task(JSONObject obj) {
      this.obj = obj;
    }

    @Override
    public void run() {
      try {
        Thread.sleep(1);
        runThis(obj);
      } catch (InterruptedException e) {
        e.printStackTrace();
      }
    }

    public void runThis(JSONObject obj) {
        //where the rows are parsed and stored in the DB, etc
    }
  }

  public static void executeCellPool(String filename) throws InterruptedException {
    // fixed pool fixed queue
    BlockingQueue<Runnable> queue = new ArrayBlockingQueue<Runnable>(300000, true);
    ThreadPoolExecutor executor = new ThreadPoolExecutor(90, 100, 1, TimeUnit.MINUTES, queue);

    DataSet ds = CommonDelimitedParser.getDataSet(filename);
    final String[] colNames = ds.getColumns();
    while (ds.next()) {
        JSONObject obj = new JSONObject();
        //some JSON object
        Task t = new Task(obj);
        executor.execute(t);
    }
  }

最佳答案

tl;dr 大队列大小可能会产生负面影响，大线程数也会产生负面影响。理想情况下，您希望消费者和生产者以相似的速度工作。

添加队列导致问题的原因是因为您正在使用占用资源的非常大的队列(这不是必需的)。通常，阻塞队列会在队列中没有剩余空间时阻塞生产者，而在队列中没有剩余对象时阻塞消费者。通过创建一个如此大的静态大小，Java 会在您几乎肯定不会使用所有空间时在内存中分配该空间。如果您的消费者成为消费者的速度太慢，那么强制您的生产者等待队列中的空间清理干净会更有效。您不需要将文件中的所有行同时存储在队列中。

线程池执行器队列在 javadoc 中讨论 here .

Bounded queues. A bounded queue (for example, an ArrayBlockingQueue) helps prevent resource exhaustion when used with finite maximumPoolSizes, but can be more difficult to tune and control. Queue sizes and maximum pool sizes may be traded off for each other: Using large queues and small pools minimizes CPU usage, OS resources, and context-switching overhead, but can lead to artificially low throughput. If tasks frequently block (for example if they are I/O bound), a system may be able to schedule time for more threads than you otherwise allow. Use of small queues generally requires larger pool sizes, which keeps CPUs busier but may encounter unacceptable scheduling overhead, which also decreases throughput.

90 的大线程，加上 300000 的超大池，很可能会使用大量内存，并导致额外的线程调度开销。我会大大放弃他们两个。我不知道你在什么硬件上运行，但由于听起来你正在编写一个 IO 密集型程序，我会尝试将你的 CPU 可以处理的线程数增加一倍，并尝试调整你的阻塞队列的大小以查看什么有效(注意:我没有对此进行研究，这是基于我运行队列和执行程序的经验。很高兴其他人提出不同的计数!)。

但值得注意的是，如果您的队列太小，execute() 方法将在无法添加到队列时抛出 RejectedExecutionException。监控队列的一种方法是在安排任务之前检查它的容量。您可以通过调用:

executor.getQueue().remainingCapacity()

不要使用 executor.getQueue() 方法以任何方式改变队列，但它可用于监控。

另一种方法是使用无界队列，例如没有定义容量的 LinkedBlockingQueue。这样，您就不需要处理队列大小。但是，如果生产者的运行速度比消费者快得多，您将再次遇到内存消耗过多的问题。

此外，kostya 是对的，JDBC 批量插入会更快。

关于java - 使用 ArrayBlockingQueue 会使进程变慢，我们在Stack Overflow上找到一个类似的问题： https://stackoverflow.com/questions/34282544/