标签: java-stream

注意:我已经在另一个SO帖子中解决了这个问题 - 在嵌套的Java 8并行流动作中使用信号量可能是DEADLOCK.这是一个错误吗？ - 但是这篇文章的标题表明问题与使用信号量有关 - 这有点分散了讨论的注意力.我正在创建这个,以强调嵌套循环可能有性能问题 - 虽然这两个问题可能是一个共同的原因(也许是因为我花了很多时间来弄清楚这个问题).(我不认为它是重复的,因为它强调另一种症状 - 但如果你只是删除它).

问题:如果嵌套两个Java 8 stream.parallel().forEach循环并且所有任务都是独立的,无状态的等等 - 除了提交到公共FJ池 - 然后在并行循环内嵌套并行循环执行得更差而不是在并行循环内嵌套顺序循环.更糟糕的是:如果同步包含内循环的操作,您将获得DEADLOCK.

演示性能问题

如果没有"同步",您仍然可以观察到性能问题.您可以在以下网址找到演示代码:http://svn.finmath.net/finmath%20experiments/trunk/src/net/finmath/experiments/concurrency/NestedParallelForEachTest.java (有关更详细的说明,请参阅JavaDoc).

我们的设置如下:我们有一个嵌套的stream.parallel().forEach().

• 内环是独立的(无状态,无干扰等 - 除了使用公共池之外)并且在最坏的情况下总共消耗1秒,即如果处理顺序.
• 外循环的一半任务在该循环之前消耗10秒.
• 在该循环之后,一半消耗10秒.
• 因此,每个线程总共消耗11秒(最坏情况).*我们有一个布尔值,允许将内部循环从parallel()切换到sequential().

现在:将24个外循环任务提交给具有并行性的池8我们期望24/8*11 =最多33秒(在8核或更好的机器上).

结果是:

• 内部顺序循环:33秒.
• 内部并行循环:> 80秒(我有92秒).

问题:你能证实这种行为吗？这是人们对框架的期望吗？(我现在更加小心,声称这是一个错误,但我个人认为这是由于ForkJoinTask的实现中的一个错误.备注:我已将此发布到并发兴趣(请参阅http:// cs.oswego.edu/pipermail/concurrency-interest/2014-May/012652.html),但到目前为止我没有得到确认).

证明了僵局

以下代码将为DEADLOCK

    // Outer loop
    IntStream.range(0,numberOfTasksInOuterLoop).parallel().forEach(i -> {
        doWork();
        synchronized(this) {
            // Inner loop
            IntStream.range(0,numberOfTasksInInnerLoop).parallel().forEach(j -> {
                doWork();
            });
        }
    });

其中numberOfTasksInOuterLoop = 24,numberOfTasksInInnerLoop = 240,outerLoopOverheadFactor = 10000和 …

Streams API中缺少的一个功能是"分区依据"转换,例如Clojure中定义的.假设我想重现Hibernate的fetch join:我想发出一个SQL SELECT语句来从结果中接收这种对象:

class Family {
   String surname;
   List<String> members;
}

我发出:

SELECT f.name, m.name 
FROM Family f JOIN Member m on m.family_id = f.id
ORDER BY f.name

我检索一个平坦的(f.name, m.name)记录流.现在我需要将其转换为Family对象流,并在其中包含其成员列表.假设我已经有了Stream<ResultRow>; 现在我需要将其转换为a Stream<List<ResultRow>>然后使用映射转换对其进行操作,将其转换为a Stream<Family>.

转换的语义如下:List只要提供的鉴别器函数保持返回相同的值,就保持将流收集到for中; 一旦值改变,发出List作为输出流的元素并开始收集新的List.

我希望能够编写这种代码(我已经有了这个resultStream方法):

Stream<ResultRow> dbStream = resultStream(queryBuilder.createQuery(
        "SELECT f.name, m.name"
      + " FROM Family f JOIN Member m on m.family_id = f.id"
      + " …

我有一个方法,每次执行它时都会生成一个对象,我需要颠倒我获取它们的顺序.所以我认为这样做的自然方式就是Stack,因为它是LIFO.

但是,Java Stack似乎不能很好地与新的Java 8流API一起使用.

如果我这样做:

   Stack<String> stack = new Stack<String>();
   stack.push("A");
   stack.push("B");
   stack.push("C");

   List<String> list = stack.stream().collect(Collectors.toList());
   System.out.println("Collected: " + list);

我得到的输出是:

Collected: [A, B, C]

为什么不以预期的LIFO顺序将它们输出到流中？这是将所有项目从堆栈清除到右(LIFO)顺序列表的正确方法吗？

我想加倍一个Stream(没有DoubleStream).意思是我从流开始并想要获得一个新流,其中旧流的每个元素都流式传输两次.所以1,2,3,4,4,5给了我们1,1,2,2,3,3,4,4,4,4,5,5.有没有这样的流操作？

我在Stream里面有Google Guava:

this.map.entrySet().stream()
.filter(entity -> !Strings.isNullOrEmpty(entity.getValue()))
.map(obj -> String.format("%s=%s", obj.getKey(), obj.getValue()))
.collect(Collectors.joining(","))

如您所见!String.isNullOrEmpty(entity),过滤器函数中有一个语句.

我不想再在项目中使用Guava了,所以我只想简单地将其替换为:

string == null || string.length() == 0;

我怎么能更优雅呢？

我想删除someMap不存在密钥的所有项目someList.看看我的代码:

someMap.keySet().stream().filter(v -> !someList.contains(v)).forEach(someMap::remove);

我收到了java.util.ConcurrentModificationException.为什么？流不是平行的.这样做最优雅的方法是什么？

我可以用这样的代码进行转换:

Object[] array = (Object[]) message.get(key);
boolean[] result = new boolean[array.length];
for (int i = 0; i < array.length; i++) {
    result[i] = (boolean) array[i];
}

但是,我认为可以使用Java 8流进行相同的转换.我开始编写这样的代码:

boolean[] =  Arrays.stream(array)
                   .map(Boolean.class::cast)
                   .map(Boolean::booleanValue)
                   .toArray()

但是这段代码不起作用.编译说

incompatible types: java.lang.Object[] cannot be converted to boolean[]

我试图了解代码的问题.我认为这map(Boolean::booleanValue)将返回我可以收集的布尔值流toArray.

我有一个列表a,我想分成几个小列表.

说出包含"aaa"的所有项目,包含"bbb"和更多谓词的所有内容.

我怎么能用java8这样做？

我看到这篇文章,但它只分成两个列表.

public void partition_list_java8() {

    Predicate<String> startWithS = p -> p.toLowerCase().startsWith("s");

    Map<Boolean, List<String>> decisionsByS = playerDecisions.stream()
            .collect(Collectors.partitioningBy(startWithS));

    logger.info(decisionsByS);

    assertTrue(decisionsByS.get(Boolean.TRUE).size() == 3);
}

我看过这篇文章,但它在Java 8之前很老了.

我有一个对象流,我想找到一个具有某个属性的最大值的那个,计算起来很昂贵.

作为一个特定的简单示例,假设我们有一个字符串列表,我们希望找到最酷的一个,给定一个coolnessIndex函数.

以下应该有效:

String coolestString = stringList
        .stream()
        .max((s1, s2) -> Integer.compare(coolnessIndex(s1), coolnessIndex(s2)))
        .orElse(null);

现在,这有两个问题.首先,假设coolnessIndex计算成本很高,这可能不会非常有效.我想这个max方法需要重复使用比较器,这反过来会coolnessIndex重复调用,最后每个字符串会多次调用它.

其次,必须提供比较器会导致代码中的一些冗余.我更喜欢这样的语法:

String coolestString = stringList
        .stream()
        .maxByAttribute(s -> coolnessIndex(s))
        .orElse(null);

但是,我无法在StreamAPI中找到匹配的方法.这让我感到惊讶,因为通过属性查找min/max似乎是一种常见的模式.我想知道是否有比使用比较器更好的方法(除了for循环).

给定一个String s和char c,我很好奇,如果存在制造的一些方法List<Integer> list从s(其中内的元素list表示的索引c内s).

一个接近但不正确的方法是:

public static List<Integer> getIndexList(String s, char c) {
    return s.chars()
            .mapToObj(i -> (char) i)
            .filter(ch -> ch == c)
            .map(s::indexOf) // Will obviously return the first index every time.
            .collect(Collectors.toList());
}

以下输入应产生以下输出:

getIndexList("Hello world!", 'l') -> [2, 3, 9]