时间戳

首页 > 解决方案 > Java并行流填充数组

java - Java并行流填充数组

问题描述

我有一个非常大的哈希图，里面充满了素数。

var mapA = new HashMap<Integer, Long>();

我需要对其进行大量计算，因此我正在使用并行流：

var res = new ArrayList<Integer();

mapA.entrySet()
        .parallelStream()
        .forEach( x -> {

            var values = mapA.entrySet()
                                    .parallelStream()
                                    .filter( /*conditions*/ )
                                    .map(y -> y.getKey())
                                    .toArray();                 

            Arrays.stream(values)
                      .parallel()
                      .sorted()
                      .forEach(val -> {

                           synchronized (this) {
                                res.add(x.getKey());
                                res.add((Integer) val);
                           }

                      });


        });

如您所见，有res一个数组超出了流的范围。我需要循环并行，否则计算可能需要几分钟和几分钟。需要这个吗？

.forEach(val -> {

    synchronized (this) {
        res.add(x.getKey());
        res.add((Integer) val);
    }

});

我添加了，synchronized因为由于流并行运行，我不希望出现竞争条件，以防 2 个或更多线程同时添加数据res。

我试图删除同步（this）并且程序仍然可以正常工作。但是我怎么能确定它总是能正常工作呢？

谢谢

如果需要，我将在此处添加整个代码：

import java.util.*;
import java.util.stream.Collectors;
import java.util.stream.IntStream;

public class DiffieHellman {

    private static final int LIMIT = 65536;

    private final long p;
    private final long g;

    public DiffieHellman(long p, long g) {
        this.p = p;
        this.g = g;
    }

    public List<Integer> tryBruteForce(long publicA, long publicB) {
        List<Integer> res = new ArrayList<Integer>();

        var mapA = new HashMap<Integer, Long>(
                IntStream
                        .rangeClosed(0, LIMIT)
                        .parallel()
                        .boxed()
                        .collect(
                                Collectors.toMap(x -> x, x -> DiffieHellmanUtils.modPow(publicB, x, p))
                        )
        );

        var mapB = new HashMap<Integer, Long>(
                IntStream
                        .rangeClosed(0, LIMIT)
                        .parallel()
                        .boxed()
                        .collect(
                                Collectors.toMap(x -> x, x -> DiffieHellmanUtils.modPow(publicB, x, p))
                        )
        );

        mapA.entrySet()
                    .parallelStream()
                    .forEach( x -> {

                        var values = mapB.entrySet()
                                        .parallelStream()
                                        .filter( y -> y.getValue().equals(x.getValue()))
                                        .map(Map.Entry::getKey)
                                        .toArray(Integer[]::new);

                        Arrays.stream(values)
                                .parallel()
                                .sorted()
                                .forEach(val -> {
                                        res.add(x.getKey());
                                        res.add((Integer) val);
                                });


                    });

        return res;
    }

}

标签： javajava-streamjava-10

解决方案

自然，您可以像其他答案指出的那样简单地使用同步集合，但这可能由于争用而性能不够，并且编写起来仍然很麻烦。

相反，您可以通过惯用的 Stream API 以稍微不同的方式解决问题。

首先，嵌套操作可以在单个流管道中完成：

mapB.entrySet()
            .parallelStream()
            .filter(y -> y.getValue().equals(x.getValue()))
            .map(y -> y.getKey())
            .sorted()
            .forEach(val -> {

                synchronized (this) {
                    res.add(x.getKey());
                    res.add((Integer) val);
                }
            });

其次，为了避免并发问题，最简单的方法是放弃命令式方法并利用 Stream API 的声明性。

为了做到这一点，人们不会手动for-each然后将add元素添加到结果中，而是让 Stream 管理它。

您在这里要做的是通过将mapAentrySet() 的每个元素替换为自定义序列来创建一个新序列：

List<Integer> res = mapA.entrySet()
      .parallelStream()
      .flatMap(x -> mapB.entrySet().stream()
         .filter(y -> y.getValue().equals(x.getValue()))
         .map(Map.Entry::getKey)
         .sorted()
         .flatMap(v -> Stream.of(x.getKey(), v)))
      .collect(Collectors.toList());

嵌套parallelStream可以省略，因为无论如何flatMap都在调用sequential()Stream。

推荐阅读