将Java While循环转换为多线程循环

Question

我有一个列表，说有基于列表内容的登录名，我正在处理其他方法，如：

while ( it.hasNext() )
{
   method1();
}

而不是等待序列，因为这些任务是独立的，所以我想并行运行它们。

我做了以下事情：

 ExecutorService service = Executors.newFixedThreadPool(5);
 final List<Future<?>> futures = new ArrayList<>();

 while ( it.hasNext() )
    {
        Future<?> future = service.submit(() -> {
                  method1(it.next());
           });
        futures.add(future);
    }

但是这个问题，它正在为一个loginName执行5次method1 ...但是我希望5个线程独立地执行method1。我在哪里做错了？

TIA！

Answer 1

尝试首先获得迭代值。

final <class> nextItem = it.next();
Future<?> future = service.submit(() -> { method1(nextItem); });
futures.add(future);

Answer 2

您的代码应该可以正常工作。我已经使用以下简单的类对其进行了测试：

public class Test {

    public static void main(String[] args) {
        List<String> loginNames = Arrays.asList("name1", "name2", "name3", "name4", "name5");
        Iterator<String> it = loginNames.iterator();

        ExecutorService service = Executors.newFixedThreadPool(5);
        final List<Future<?>> futures = new ArrayList<>();

        while (it.hasNext()) {
            Future<?> future = service.submit(() -> {
                method1(it.next());
            });
            futures.add(future);
        }
    }

    public static void method1(String v) {
        System.out.println(v);
    }
}

并打印所有名称，例如：

name1
name3
name2
name4
name5

顺序不同，这意味着它将并行打印它们。

但是我建议您使用Java 8 .parallelStream()使其更简​​单：

public class Test {

    public static void main(String[] args) {
        List<String> loginNames = Arrays.asList("name1", "name2", "name3", "name4", "name5");
        loginNames.parallelStream().forEach(Test::method1);
    }

    public static void method1(String v) {
        System.out.println(v);
    }
}

Answer 3

您的方法不是线程安全的。每次迭代都会调用it.hasNext()，因此您可以相信获得了预期的结果，但是Iterator并不是线程安全的。因此，如果同时执行多个Callable，则使用Iterator会产生不可预测的结果。

使用线程或Callable当然可以利用并行性，但是在使用迭代器时，您将需要添加显式同步。使用Callable进行显式同步在某种程度上令人讨厌。

作为替代，您可以使用确保线程安全的集合，例如CopyOnWriteArrayList。例如，将原始列表包装在CopyOnWriteArrayList复制构造函数中，例如：

List<Foo> copyList = new CopyOnWriteArrayList<>(list);

或者，您也可以更简单地使用并行流，这将使您的代码不再那么冗长，但是它对于后台线程池的可配置性较差（在某些情况下，这很重要）：

list.parallelStream().forEachOrdered(this::method1);

Answer 4

我在哪里做错了？

您有一个正在执行此操作的线程：

while ( it.hasNext() ) {
    Future<?> future = service.submit(...task...);
    futures.add(future);
}

该线程一直在调用it.hasNext()，但从不调用it.next()。

您有other个线程（线程池的工作程序）调用了it.next()，但这些线程独立于循环运行。如果循环将一个任务提交给执行者服务并在列表中添加未来的时间少于执行一个任务所花费的时间，则该循环将领先执行者服务，并且将提交更多的执行者服务。任务比迭代器中的字符串数多。