groovy 中的并行流答案

【问题标题】：parallel Stream in groovygroovy 中的并行流
【发布时间】：2022-11-05 10:14:36
【问题描述】：

我正在尝试在 groovy 中处理两个具有并行流的地图。我有两个 <object, ArrayList> 类型的 HashMap 类型的映射。地图A和地图B。我需要在 mapA 上与 mapB 并行流，以检查 MapA 的密钥是否存在于 MapB 中。如果存在，则比较 arraylist 中的值。

def Map = [:].withDefault {key -> return []}
-
-
-
//populating two maps
    def jsonSlurper = new JsonSlurper()

    while ((inputLine = reader.readLine()) != null) {
        if (inputLine.startsWith('{"k"')) {
            def json =jsonSlurper.parseText(inputLine)
            Map.put(json.key[3],
                    [json.key[4],json.key[5]])
        }
    }

// comparing to map to check if key exists, if yes then compare value[0] of mapA to mapB's Value[0].. and then value[1] and so on. 
def compareDatastore = { mapA,mapB ->
mapA.entrySet().parallelStream().with {
    **it.forEach(entry->{**
        if(mapB.containsKey(entry.getKey())){
            if(entry.getValue().get(0)!=mapB.get(entry.getKey()).get(0) || entry.getValue().get(1)!=mapB.get(entry.getKey()).get(1))
                println "noMatch"
        }else{
            println "notFound"
        }
    })
}
}

我怎样才能做得更好？

地图中的样本值为

key=1245,value=[a,b]
key=1234,value=[b,a]

there will always be only two value in arraylist.

在上述代码中的 foreach 行出现以下错误。

Caught: java.lang.VerifyError: Bad local variable type
Exception Details:
  Location:
    scripts/smething$_run_closure6$_closure8$_closure9.doCall(Ljava/lang/Object;)Ljava/lang/Object; @155: aload_3
  Reason:
    Type top (current frame, locals[3]) is not assignable to reference type
  Current Frame:

非常感谢任何帮助！

【问题讨论】：

为什么需要并行处理？
我将在内存中获取数千条记录并进行比较。我将在两个映射中流式传输两个数据源，然后在内存中比较它们。早些时候，我正在流式传输一个数据存储，并在多线程环境中为每个引脚进行 db 调用。
如果没有样本输入，这很难解决。另外，我认为您不需要并行处理
另外，为什么需要使用流？
嗨，蒂姆，我提供了里面的示例值，这些值将在地图内。我有两张地图，里面有相似的数据结构。我需要将这两个地图相互比较。将检查 MapB 是否包含 MapA 中存在的“键”。 MapA 和 MapB 由数据库填充，并一次提取数千条记录并进行内存比较，通过这样做，我减少了 50% 的网络调用。

标签： java groovy stream hashmap

【解决方案1】：

我对输入映射值进行了硬编码，但下面的代码可能符合您的预期：

def mapA = [
    'key1': ['1', '2'],
    'key2': ['3', '4'],
    'key4': ['5', '6']
]

def mapB = [
    'key1': ['1', '2'],
    'key3': ['3', '4'],
    'key4': ['5', '7']
]

mapA.entrySet().parallelStream().forEach { entry ->
    if (mapB.containsKey(entry.key)) {
        if (entry.value != mapB[entry.key]) {
            println "'${entry.key}' key doesn't have matching value"
        } else {
            println "'${entry.key}' key has matching value"
        }
    } else {
        println "'${entry.key}' key not found"
    }
}

执行时，它会打印以下行：

'key2' key not found
'key4' key doesn't have matching value
'key1' key has matching value

由于并行处理，日志行的顺序将在执行之间发生变化

【讨论】：