如何提交列表 >到ThreadPoolExecutor,每个线程将选择一个LinkedBlockingQueue并并行执行

问题描述 投票:0回答:2

我将Long类型的LinkedBlockingQueue列表提交给ThreadPoolExecutor,条件应该是每个线程选择long的LinkedBlockingQueue并且并行执行

这是我的方法逻辑

public void doParallelProcess() {

    List<LinkedBlockingQueue<Long>> linkedBlockingQueueList = splitListtoBlockingQueues();
    ThreadPoolExecutor executor = new ThreadPoolExecutor(1, linkedBlockingQueueList.size(), 0L,
            TimeUnit.MILLISECONDS, new LinkedBlockingQueue<Runnable>(), Executors.defaultThreadFactory());
    Long initial = System.currentTimeMillis();
    try {

        System.out.println("linkedBlockingQueueList begin size is " + linkedBlockingQueueList.size() + "is empty"
                + linkedBlockingQueueList.isEmpty());

        while (true) {
            linkedBlockingQueueList.parallelStream().parallel().filter(q -> !q.isEmpty()).forEach(queue -> {
                Long id = queue.poll();
                MyTestRunnable runnab = new MyTestRunnable(id);
                executor.execute(runnab);
                System.out.println("Task Count: " + executor.getTaskCount() + ", Completed Task Count: "
                        + executor.getCompletedTaskCount() + ", Active Task Count: " + executor.getActiveCount());
            });

            System.out.println("linkedBlockingQueueList end size is " + linkedBlockingQueueList.size() + "is empty"
                    + linkedBlockingQueueList.isEmpty());

            System.out.println("executor service " + executor);

            if (executor.getCompletedTaskCount() == (long) mainList.size()) {
                break;
            }

            while (executor.getActiveCount() != 0) {
                System.out.println("Task Count: " + executor.getTaskCount() + ", Completed Task Count: "
                        + executor.getCompletedTaskCount() + ", Active Task Count: " + executor.getActiveCount());
                Thread.sleep(1000L);
            }

        }
    } catch (Exception e) {
    } finally {
        executor.shutdown();
        while (!executor.isTerminated()) {
        }
    }
} `

如何将LinkedBlockingQueue列表提交给单个线程示例:

  1. List<LinkedBlockingQueue<Long>>每个LinkedBlockingQueue包含50个队列数据
  2. List<LinkedBlockingQueue<Long>>的大小是50
  3. 每个线程应该选择一个LinkedBlockingQueue<Long>并执行50个队列任务。
java multithreading threadpool executorservice java-threads
2个回答
1
投票

ExecutorService的输入是RunnableCallable。您提交的任何任务都需要实现这两个接口之一。如果你想向线程池提交一堆任务并等到它们全部完成,那么你可以使用invokeAll方法并循环生成的Futures,在每个上调用get:请参阅此信息性answer到类似的问题。

但是,您不需要将输入任务批量分组。你还不希望执行程序服务有空闲线程,而还有剩下的工作要做!您希望它能够在资源释放后立即获取下一个任务,并且以这种方式进行批处理与此相反。你的代码是这样做的:

while non-empty input lists exist {
    for each non-empty input list L {
        t = new Runnable(L.pop())
        executor.submit(t)
    }
    while (executor.hasTasks()) {
        wait
    }
}

一旦其中一个任务完成,该线程就可以自由地继续进行其他工作。但它不会因为你等到所有N个任务完成之后再提交。使用invokeAll一次性提交它们,让执行程序服务完成它的构建。


0
投票

Executors类是线程池的主要入口:

    ExecutorService executor = Executors.newCachedThreadPool();
    linkedBlockingQueueList.forEach(queue -> executor.submit(() -> { /* process queue */ }));

如果你想自己创建一个ThreadPoolExecutor - 它确实可以让你更好地控制配置 - 至少有两种方法你可以指定一个默认的线程工厂:

  1. 省略线程工厂参数: ThreadPoolExecutor executor = new ThreadPoolExecutor(1, linkedBlockingQueueList.size(), 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<Runnable>());
  2. 再次使用Executors类获取默认的线程工厂: ThreadPoolExecutor executor = new ThreadPoolExecutor(1, linkedBlockingQueueList.size(), 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<Runnable>(), Executors.defaultThreadFactory());
© www.soinside.com 2019 - 2024. All rights reserved.