【问题标题】:Process list of 'N' items with multiple threads处理具有多个线程的“N”项的列表
【发布时间】:2015-08-19 05:17:05
【问题描述】:

我有ListN 项目,我想在固定数量的threads 之间按顺序划分这个List

我的意思是,我想将1 to N/4传递给第一个threadN/4 + 1 to N/2传递给第二个线程,将N/2+1 to N传递给第三个thread,现在一旦所有threads都完成了他们的工作,我想通知主thread 发送一些消息,所有处理都已完成。

目前为止我所做的是实现了ExecutorService

我做了这样的事情

ExecutorService threadPool = Executors.newFixedThreadPool(Number_of_threads); 
                //List of items List
                List <items>itemList = getList(); 
                  for (int i = 0 i < Number_of_threads ;i++ ) { 
                    //how to divide list here sequentially and pass it to some processor while will process those items.
                    Runnable processor = new Processor(Start, End)
                    executor.execute(process);
                   }
                  if(executor.isTerminated()){
                    logger.info("All threads completed");
                  }
  • 如何将列表分成连续的块?
  • 有没有更好的方法来实现这样的功能?

【问题讨论】:

标签: java multithreading concurrency threadpoolexecutor


【解决方案1】:

如果您想要让所有线程尽快完成处理并且项目数量不是巨大,那么只需将每个项目发布一个RunnablenewFixedThreadPool(NUMBER_OF_THREADS): p>

    ExecutorService exec = Executors.newFixedThreadPool(NUMBER_OF_THREADS);
    List<Future<?>> futures = new ArrayList<Future<?>>(NUMBER_OF_ITEMS);
    for (Item item : getItems()) {
        futures.add(exec.submit(new Processor(item)));
    }
    for (Future<?> f : futures) {
        f.get(); // wait for a processor to complete
    }
    logger.info("all items processed");

如果您真的想给每个线程一个列表的连续部分(但仍然希望它们尽快完成,并且还希望处理每个项目花费大约相同的时间),然后尽可能“均匀地”拆分项目,以便每个线程的最大项目数与最小数量的差异不超过一个(例如:14 项目,4 线程,那么你想要拆分是[4,4,3,3],而不是例如[3,3,3,5])。为此,您的代码将是例如

    ExecutorService exec = Executors.newFixedThreadPool(NUMBER_OF_THREADS);
    List<Item> items = getItems();
    int minItemsPerThread = NUMBER_OF_ITEMS / NUMBER_OF_THREADS;
    int maxItemsPerThread = minItemsPerThread + 1;
    int threadsWithMaxItems = NUMBER_OF_ITEMS - NUMBER_OF_THREADS * minItemsPerThread;
    int start = 0;
    List<Future<?>> futures = new ArrayList<Future<?>>(NUMBER_OF_ITEMS);
    for (int i = 0; i < NUMBER_OF_THREADS; i++) {
        int itemsCount = (i < threadsWithMaxItems ? maxItemsPerThread : minItemsPerThread);
        int end = start + itemsCount;
        Runnable r = new Processor(items.subList(start, end));
        futures.add(exec.submit(r));
        start = end;
    }
    for (Future<?> f : futures) {
        f.get();
    }
    logger.info("all items processed");

【讨论】:

    猜你喜欢
    • 1970-01-01
    • 1970-01-01
    • 1970-01-01
    • 2018-03-11
    • 1970-01-01
    • 1970-01-01
    • 1970-01-01
    • 2021-11-09
    • 2021-10-27
    相关资源
    最近更新 更多