ParallelStream 收集到设置？

Question

我有一个程序可以从使用parallelStream() 中受益匪浅（大型数据集，其中涉及一些映射和过滤方案，但不依赖于外部变量/同步），但必须作为一个集合收集。

我对并行流有些陌生（这是我第一次使用它们），并尝试使用以下代码却发现这导致了非并发后端的并发修改和幕后的死锁情况.

此映射尝试使用 Linux 本机命令 sudo blockdev --getsize64 unmounted_device_here 获取未安装磁盘的文件大小（我不知道 Java 是否可以在 Linux 上获取未安装磁盘的完整大小，所以我'我只是使用本机方法，因为无论如何这只会在 Linux 系统上发布）

映射方法（死锁）：

var mountPath = Paths.get("/dev");
            //Do NVME Drives First
            var list = new ArrayList<Path>(10);
            //Looks like nvme1n1
            //For reasons beyond my understanding replacing [0-9] with \\d does not work here
            try (var directoryStream = Files.newDirectoryStream(mountPath, "nvme[0-9]n[0-9]")) {
                for (Path path : directoryStream) {
                    list.add(path);
                }
            }
//Map to DrivePacket (path, long), note that blockdev return bytes -> GB
var nvmePackets = list.parallelStream().map((drive) -> new DrivePacket(drive,
                    (Long.parseLong(runCommand("sudo", "blockdev", "--getsize64", drive.toAbsolutePath().toString())) / (1024 * 1024 * 1024))))
                    .collect(Collectors.toSet());

IOUtils 来自 Apache 实用程序类：

        <dependency>
            <groupId>org.apache.commons</groupId>
            <artifactId>commons-lang3</artifactId>
            <version>3.12.0</version>
        </dependency>

runCommand（执行本机调用）：

public static String runCommand(String... command) {
        try {
            if (DEBUG_MODE) {
                systemMessage("Running Command: " + Arrays.asList(command).stream().collect(Collectors.joining(" ")));
            }
            var builder = new ProcessBuilder(command);
            var result = IOUtils.toString(builder.start().getInputStream(), StandardCharsets.UTF_8).replaceAll("\n", "");
            if (DEBUG_MODE) {
                System.out.println("Result: " + result);
            }
            return result;
        } catch (IOException ex) {
            throw new IllegalStateException(ex);

        }
    }

DrivePacket 类：

   /**
     * A record of the relevant information for a drive
     *
     * Path is the fully qualified /dev/DRIVE path
     */
    public record DrivePacket(Path drivePath, long driveSize) {}

既然操作受益于并发，有没有办法使用parallelStream来做到这一点？还是我必须使用其他技术？

它总是挂在执行这行代码的停止处，并在我使用调试器时在ForkJoinTask.javaexternalAwaitDone(); 处永远等待。

很遗憾，我找不到与 toConcurrentMap() 类似的 Set。

我怎样才能避免这种死锁，同时仍然获得计算的并行性并让最终结果成为一个集合？

系统：JDK 16

编辑 0：更新了可重现性代码

鉴于映射代码调用不共享数据的子例程，我不确定为什么这会导致死锁情况。

Answer 1

也许您可以将其收集到ConcurrentMap，然后获取keySet。 （假设为映射对象定义了equals、hashcode方法）：

list.parallelStream()
    .map(x -> complicated_mapping_here)
    .collect(Collectors.toConcurrentMap(Function.identity(),
                                        x -> Boolean.TRUE /*dummy value*/ ))
    .keySet();