Streamを途中で分岐させたい

Question

stream を途中で分岐させたいです
例えば、得点が60点以上だった場合に1ポイント、80点以上だった場合に2ポイント、90点以上は 3 + (n - 90) ポイント付与する条件で、ポイントの合計が欲しい場合。
streamを分岐できれば簡単に書けると思うのですが、やりようはないでしょうか

コードのイメージとしてはこんなかんじです

Stream<Integer> stream = Stream.of(1, 2, 3, 60, 65, 85, 90, 92, 95);
Integer sum = stream.flatMap(
    fork(
      branch(n -> n < 60).map(n -> 0),
      branch(n -> n < 80).map(n -> 1),
      branch(n -> n < 90).map(n -> 2),
      branch().map(n -> 3 + (90 - n))
    )
  )
  .mapToInt(Integer::valueOf)
  .sum();

ちなみに、途中で終端処理を挟まない方法を模索中です

Answer 1

実現したいのは、下記のようなデーターフローでしょうか？

[Source] --> [FilterX] --> [Split] --> [FilterA] --> [SinkA]
                              +------> [FilterB] --> [SinkB]

残念ながら、Java8のストリーム(java.util.stream.Stream)は、汎用のデータフロー・パイプライン処理のための仕組みではありません。出力先が2個以上あるような「分岐」は表現不可能です。

1つのストリームは、必ず「1つのソース」＋「0個以上の中間操作」＋「1つの終端操作」から構成されます。

Streamが何を目的としたものであるかや、その内部的な動作については 社内Java8勉強会 ラムダ式とストリームAPI が参考になるかと思います。

Comments

[kakusuke]

なるほど、原理的に無理なんですね・・・
理解しました、ありがとうございます。

Answer 2

グルーピング化して、それぞれ計算するようにしてみるぐらいしか方法はないかなと思います。
並列化もしているので、時間がかかる処理もマルチスレッドでよしなにやってくれるはずです。
こんな感じでいかがでしょうか？
(もうちょっとうまいか書き方はあるかもしれません)

import java.util.stream.*;

public class SumPoint2 {
  public static void main(String[] args) {
    Stream<Integer> stream = Stream.of(1, 2, 3, 60, 65, 85, 90, 92, 95);
    Integer sum = stream.collect(Collectors.groupingByConcurrent((n) -> {
        if (n >= 90) {
          return "A";
        } else if (n >= 80) {
          return "B";
        } else if (n >= 60) {
          return "C";
        } else {
          return "Z";
        }
      })).entrySet().parallelStream().mapToInt((e) -> {
        switch(e.getKey()) {
          case "A":
            return e.getValue().stream().mapToInt(n -> 3 + (n - 90)).sum();
          case "B":
            return 2 * e.getValue().size();
          case "C":
            return e.getValue().size();
        }
        return 0;
      }).sum();

    System.out.println(sum); // => 20
  }
}

---

以下は普通にする、修正前の回答です。

Streamを使うのであれば、処理を分岐させるのではなく、
各ポイントにマッピングすると考えた方がいいです。
下のような感じはいかがでしょうか？

import java.util.stream.*;

public class SumPoint {
  public static void main(String[] args) {
    IntStream stream = IntStream.of(1, 2, 3, 60, 65, 85, 90, 92, 95);
    Integer sum = stream
      .map((n) -> {
        if (n >= 90) {
          return 3 + (n - 90);
        } else if (n >= 80) {
          return 2;
        } else if (n >= 60) {
          return 1;
        } else {
          return 0;
        }
      })
      .sum();
    System.out.println(sum); // => 20
  }
}

Comments

[kakusuke]

ご回答ありがとうございます

ただすみません、あくまで例であって、やりたいことはstreamの分岐なんです
集合の中の部分集合毎単位で処理を記述したいんです
回答していただいた方法だと、要素単位での記述になるので、すみませんが求めている答えではないです

[raccy]

いまいちしたいことのイメージがつかみ切れてませんが、回答を修正してみました。
集合毎単位で処理・・・していると思います。

[kakusuke]

説明が足りずすみません、、、

stream で書いてる最中に、まさに2つめのように一旦終端処理を書いたあとentrySetで再度ストリームを作るみたいな書き方になってしまったコードがありました
このままだと可読性とメモリ効率が悪いので、ストリームの中間処理としてサブストリームを作り、それぞれのサブストリームの中で集計し、本流に合流できればスマートだと考えた次第です

[raccy]

LINQのGroupByみたいにストリームを直接グルーピングできたら
まだ少しはましだったのかも知れませんけど、Java8にはないみたいですね。

Answer 3

集合の中の部分集合毎単位で処理を記述したい

ということであれば元のストリームから目的の要素を(filterで)抜き出したストリームを作ればいいのでは？

IntStream grade0 = stream.filter((n) -> EvaluationRule.classify(n) == 0);

合計などが欲しくなったら分けたストリームを統合すればいいでしょう。

---

思いっきりC#のノリで書いてしまいましたが、
javaのStreamは使い回しができないので
一々ストリーム元からストリームを生成する必要がありました。

int[] source = { 1,2,3,4,5};
IntStream s1 = Arrays.stream(source).filter(i -> i < 3).map(i -> 1);
IntStream s2 = Arrays.stream(source).filter(i -> i >= 3).map(i -> 2);
IntStream s12 = IntStream.concat(s1, s2);

Comments

[kakusuke]

うーん、どうやら一度処理したストリームは再度処理できないようです・・・
このままでは分岐は実現できないですね・・・

IntStream stream = IntStream.of(1,2,3,4,5);
IntStream s1 = stream.filter(i -> i < 3).map(i -> 1);
IntStream s2 = stream.filter(i -> i >= 3).map(i -> 2);
System.out.println(IntStream.concat(s1, s2).sum()); // 10 を期待

Exception in thread "main" java.lang.IllegalStateException: stream has already been operated upon or closed
at java.util.stream.AbstractPipeline.<init>(AbstractPipeline.java:203)
at java.util.stream.IntPipeline.<init>(IntPipeline.java:91)
at java.util.stream.IntPipeline$StatelessOp.<init>(IntPipeline.java:592)
at java.util.stream.IntPipeline$9.<init>(IntPipeline.java:332)
at java.util.stream.IntPipeline.filter(IntPipeline.java:331)
at Main.main(Main.java:7)