When to Stream, Collect, and Gather

Slide navigation: Forward with space bar, →, or PgDn. Backwards with ← or PgUp.

About the Speaker

• Cay Horstmann
• Author of Core Java (14 editions since 1996), Java/JavaScript/Scala for the Impatient, etc.
• Java Champion
• https://horstmann.com

Why Use Streams?

• Can be parallelized
• Can be more efficient (no intermediate collections)
• Declarative (“what, not how”)

  txns.stream()
      .filter(t -> t.getBuyer().getAge() >= 65)
      .map(Txn::getSeller)
      .distinct()
      .sorted(comparing(Seller::getName))
      .map(Seller::getName)
      .forEach(System.out::println);
  

• instead of

  Set<Seller> sellers = new HashSet<>();
  for (Txn t : txns) {
      if (t.getBuyer().getAge() >= 65)
          sellers.add(t.getSeller());
  }
  List<Seller> sorted = new ArrayList<>(sellers);
  Collections.sort(sorted, new Comparator<Seller>() {
      public int compare(Seller a, Seller b) {
          return a.getName().compareTo(b.getName());
      }
  });
  for (Seller s : sorted)
      System.out.println(s.getName());
  

  (Source)

Why Not Always Streams?

  static int numberOfBrickWithChildrenWithAtLeastTwoParents(String input) {
    var grid = new HashMap<Pos, Level>();
    return input.lines()
        .map(l -> {
          var array = Arrays.stream(l.split(",|~")).mapToInt(Integer::parseInt).toArray();
          return new Brick(array[0], array[3], array[1], array[4], array[2], array[5]);
        })
        .collect(groupingBy(Brick::z1, TreeMap::new, toList()))
        .values().stream()
        .flatMap(List::stream)
        .map(brick -> {
          var onTop = brick.positions()
              .map(pos -> grid.getOrDefault(pos, new Level(null, 0)))
              .collect(groupingBy(Level::z, TreeMap::new, flatMapping(l -> Stream.ofNullable(l.brick), toSet())))
              .lastEntry();
          var level =  new Level(brick, 1 + brick.z2 - brick.z1 + onTop.getKey());
          brick.positions().forEach(pos -> grid.put(pos, level));
          return new Pair<>(brick, onTop.getValue());
        })
        .collect(teeing(
            toMap(Pair::u, pair -> pair.v().size()),
            flatMapping(pair -> pair.v().stream().map(b -> new Pair<>(b, pair.u())),
                groupingBy(Pair::u, mapping(Pair::v, toList()))),
            (parentMap, childrenMap) ->
                (int) parentMap.keySet().stream()
                    .filter(parent -> childrenMap.getOrDefault(parent, List.of()).stream()
                        .allMatch(child -> parentMap.get(child) > 1))
                    .count()
        ));
  }

Max and Min

• Classic max/min:

  T largest = Collections.max(collection, comparator);
  

• Stream version:

  T largest = collection.stream().max(comparator).orElse(default);
  

• Stream version behaves better with empty collection
• Stream version works with array
• Verdict: In 2025, use streams for max/min

Finding, Counting, Collecting Matches

• Can find element in a list with indexOf
• But only if match means equals
• Otherwise loop is messy:

  boolean found = false;
  int index = 0;
  while (index < list.length() && !found) {
     if (list.get(index) is a match) found = true;
     else index++;
  }
  if (found) ...
  

• Stream solution is clean:

  Optional<T> result = list.stream().filter(e -> e is a match).findFirst();
  

• Counting, collecting matches also simpler with streams:

  long count = list.stream().filter(e -> e is a match).count();
  List<T> matches = list.stream().filter(e -> e is a match).toList();
  

Nothing to See Here...

• Advent of Code 2023 Problem 1: Add the numbers made from the first and last digit in each input line:

  1abc2
  pqr3stu8vwx
  a1b2c3d4e5f
  

• Classic loop solution:

  Pattern p = Pattern.compile("^[^\\d]*(\\d).*(\\d)[^\\d]*$");
  List<String> lines = Files.readAllLines(path);
  int sum = 0;
  for (String line : lines) {
      Matcher m = p.matcher(line);
      if (m.matches()) {
          int firstLast = Integer.parseInt(m.group(1) + m.group(2));
          sum += firstLast;
      }
  }
  

• Stream solution:

  int sum = Files.lines(path)
      .map(line -> p.matcher(line))
      .filter(m -> m.matches())
      .mapToInt(m -> Integer.parseInt(m.group(1) + m.group(2)))
      .sum();
  

• Stream solution sends crucial message—move along, nothing to see here...

Method Expressions

• Previous solution with method expressions:

  int sum = lines.stream()
      .map(p::matcher)
      .filter(Matcher::matches)
      .mapToInt(m -> Integer.parseInt(m.group(1) + m.group(2)))
      .sum();
  

• Verdict: Use them—the method name is clearer than the method body.
• Helpful table

  Variant	Lambda expression	Method expression
  Class::staticMethod	x -> Pattern.compile(x)	Pattern::compile
  Class::instanceMethod	x -> x.matches()	Matcher::matches
  object::instanceMethod	x -> p.matcher(x)	p::matcher
  Out of luck	x -> x.substring(k)

For Each

• Oracle Tutorial by Venkat Subramaniam:

  for (int i = 0; i < 5; i++) {
    System.out.println(i);
  }
  

  ⇒

  IntStream.range(0, 5)
    .forEach(i -> System.out.println(i));
  

• Steps:

  for (int i = 0; i < 15; i += 3) ...
  
  

  ⇒

  IntStream.iterate(0, i -> i < 15, i -> i + 3).forEach(...)
  

• Nested loop:

  for (int i = 0; i < 5; i++)
     for (int j = 0; j < 10; j++)
        System.out.println(i + " " + j);
  

  ⇒

  IntStream.range(0, 5).forEach(i ->
      IntStream.range(0, 10).forEach(j ->
          System.out.println(i + " " + j)))
  

• Not easier to read
• Can’t mutate variables in lambda
• Not really “functional”
• Verdict: Avoid forEach

Streams are Everywhere in the Java API

• Lines: String.lines(), Files.lines(path), HttpResponse.BodyHandlers.ofLines()
• Tokens: Scanner.tokens(), Pattern.splitAsStream(input)
• Matches: Matcher.results(), Scanner.findAll(pattern)
• Files: Files.list(path), Files.walk(path), Files.find(...)
• Process handles: Process.children(), ProcessHandle.allProcesses(), ...
• Dates: LocalDate.datesUntil(endExclusive)
• Random numbers: RandomGenerator.ints(), RandomGenerator.ints(from, to)
• Code points and code units: String.codePoints(), String.chars()
• Miscellaneous: BitSet.stream(), Locale.availableLocales(), Currency.availableCurrencies(), TimeZone.availableIDs(), NetworkInterface.inetAddresses(), DriverManager.drivers()
• More attractive than returning a collection or iterator
• Particularly for lazy evaluation
• Nudges you into the stream API
• You always have the option to call toList() or limit(n).toList()
• Or, ugh, forEach
• Be sure to close resources:

  try (Stream<String> s = Files.lines(path)) {
      s.forEach(...);
  }
  

Complex Lambdas

• Advent of Code 2023 Day 2: Given upper limit for each round (12 red, 13 green, and 14 blue) and

  Game 1: 3 blue, 4 red; 1 red, 2 green, 6 blue; 2 green
  Game 2: 1 blue, 2 green; 3 green, 4 blue, 1 red; 1 green, 1 blue
  Game 3: 8 green, 6 blue, 20 red; 5 blue, 4 red, 13 green; 5 green, 1 red
  ...
  

  add the IDs of valid games
• Rémi’s ingenious solution:

  var expected = Map.of("red", 12, "green", 13, "blue", 14);
  var PATTERN = Pattern.compile("Game ([0-9]+)|([0-9]+) (red|green|blue)");
  var result = input.lines()
      .mapToInt(line -> {
          var matcher = PATTERN.matcher(line);
          matcher.find();
          var gameId = parseInt(matcher.group(1));
          return matcher.results().allMatch(result ->
              parseInt(result.group(2)) <= expected.get(result.group(3))) ? gameId : 0;
      })
      .sum();
  

• Avoid lambdas with complex bodies
• Hard to read, hard to debug
• Add types (game, round)
• Break into smaller steps:

  .map(Game::parse)
  .filter(Game::isValid)
  .mapToInt(Game::id)
  .sum()
  

Nested Streams

• In general, be wary of streams in streams:

  var result = input.lines()
      .mapToInt(line -> {
          var matcher = PATTERN.matcher(line);
          matcher.find();
          var gameId = parseInt(matcher.group(1));
          return matcher.results().allMatch(result -> parseInt(result.group(2)) <= expected.get(result.group(3))) ? gameId : 0;
      })
      .sum();

• Idiomatic exceptions with flatMap
• Skipping empty optionals with flatMap and stream:

  var commandLines = ProcessHandle.allProcesses()
      .flatMap(p -> p.info().commandLine().stream())
      .toList()
  

• Rows and columns with flatMap and map:

  var result = IntStream.range(0, 5).boxed().flatMap(i ->
          IntStream.range(0, 10).mapToObj(j ->
              "" + i + " " + j))
      .toList(); // ["0 0", "0 1", ..., "4 9"]
  

Ad-Hoc Records

• Maurice Naftalin and José Paumard, Devoxx Belgium 2024

  record IndexWord(int index, String value) {}
  
  IndexWord splitStream(String line, String delimiter) {
      var pattern = Pattern.compile(delimiter);
      var words = Pattern.split(line);
      return IntStream.range(0, words.length)
          .filter(i -> words[i].length() == 3)
          .mapToObj(i -> new IndexWord(i, words[i]))
          .findFirst().orElseThrow();
  }
  

• Ugh—a whole new type for this use case
• Consider

  record IndexedValue<T>(int index, T value) {}
  

• Or use Map.Entry<K, V> as poor man’s pair
• Ad-hoc records are fine when they are local
• Why not use

  new Scanner(line).useDelimiter(delimiter).tokens()
  

• Hold that thought—need zipWithIndex

Collectors

• Terminal stream operation collect(Collector)
• Collectors provides 40 methods that yield implementations
• Can also roll your own (uncommon)
• Can be easy:

  Map<Integer, String> idToName = people.collect(
      Collectors.toMap(Person::id, Person::name));
      // Only use when the keys are distinct
  

• Or baffling:

  Map.Entry<List<String>, Double> result = // Map.Entry is poor man’s Pair
      Stream.of(cities)
      .filter(c -> c.state().equals("NV"))
      .collect(teeing(
          mapping(City::name, toList()), // First downstream collector
          averagingDouble(City::population), // Second downstream collector
          Map::entry)); // Combining function
  

• By far the most common case: Collecting to a map

  Map<String, List<Person>> nameToPerson = people.collect(groupingBy(Person::name));
  

• Tip: import static java.util.stream.Collectors.*;

Counting Groups

• Map of counts:

  Map<String, Long> counts = Stream.of(words)
      .collect(groupingBy(s -> s, counting()));
  

• Clearer than the classic loop:

  for (String word : words)
      counts.merge(word, 1L, Long::sum);
  

• Why not Function::identity instead of s -> s?
• Are you kidding? That’s fourteen more characters!

• |  no suitable method found for groupingBy(Function::identity,java.util.stream.Collector<java.lang.Object,capture#9 of ?,java.lang.Long>)
  |      method java.util.stream.Collectors.<T,K,D,A,M>groupingBy(java.util.function.Function<? super T,? extends K>,java.util.function.Supplier<M>,java.util.stream.Collector<? super T,A,D>) is not applicable
  |        (cannot infer type-variable(s) T,K,D,A,M
  |          (actual and formal argument lists differ in length))
  |      method java.util.stream.Collectors.<T,K,A,D>groupingBy(java.util.function.Function<? super T,? extends K>,java.util.stream.Collector<? super T,A,D>) is not applicable
  |        (cannot infer type-variable(s) T,K,A,D,capture#10 of ?,T
  |          (argument mismatch; unexpected static method <T>identity() found in unbound lookup))
  |      method java.util.stream.Collectors.<T,K>groupingBy(java.util.function.Function<? super T,? extends K>) is not applicable
  |        (cannot infer type-variable(s) T,K
  |          (actual and formal argument lists differ in length))
  |      .collect(groupingBy(Function::identity, counting()));
  |               ^--------^
  
  

• Oops—that should be Function.identity()

Gatherers

• JEP 485 introduces gather for intermediate operations
• Five built-in gatherers
• Fixed and sliding window:

  IntStream.range(0, 10).boxed().gather(Gatherers.windowFixed(4)).toList()
      // [[0, 1, 2, 3], [4, 5, 6, 7], [8, 9]]
  IntStream.range(0, 10).boxed().gather(Gatherers.windowSliding(4)).toList()
      // [[0, 1, 2, 3], [1, 2, 3, 4], [2, 3, 4, 5], [3, 4, 5, 6], [4, 5, 6, 7], [5, 6, 7, 8], [6, 7, 8, 9]]
  

• Useful to compare/combine/group adjacent elements
• Scan and fold:

  Integer[] digits = { 1, 7, 2, 9 };
  Stream.of(digits).gather(Gatherers.scan(() -> 0, (x, y) -> x * 10 + y)) // [1, 17, 172, 1729]
  Stream.of(digits).gather(Gatherers.fold(() -> 0, (x, y) -> x * 10 + y)) // [1729]
  

• mapConcurrent—see below
• You can roll your own:

  static <T> Gatherer<T, ?, Map.Entry<Long, T> zipWithIndex() {
      class Index { long index = 0; }
      return Gatherer.ofSequential(
              Index::new,
              Gatherer.Integrator.ofGreedy((state, element, downstream) ->
                      return downstream.push(Map.Entry(state.index++, element))));
  }
  

• Additional gatherers: gatherers4j, more-gatherers, packrat

Checked Exceptions

• JavaSpecialists newsletter #314:

  Files.walkFileTree(source, new SimpleFileVisitor<>() {
      public FileVisitResult visitFile(Path file, BasicFileAttributes attrs) throws IOException {
          if (Files.isSymbolicLink(file)) {
              Path targetLink = target.resolve(source.relativize(file));
              Files.createDirectories(targetLink.getParent());
              Files.copy(file, targetLink, StandardCopyOption.REPLACE_EXISTING, LinkOption.NOFOLLOW_LINKS);
          }
          return FileVisitResult.CONTINUE;
  }});

• I chide Heinz for not using modern I/O. So much simpler. Except...exceptions...

      try (Stream<Path> walk = Files.walk(source)) {
          walk.forEach(file -> {
              if (Files.isSymbolicLink(file)) {
                  Path targetLink =target.resolve(source.relativize(file));
                  try {
                      Files.createDirectories(targetLink.getParent());
                      Files.copy(file, targetLink, StandardCopyOption.REPLACE_EXISTING, LinkOption.NOFOLLOW_LINKS);
                  } catch (IOException e) {
                      throw new UncheckedIOException(e);
                  }
              }
          });
      }
      catch (UncheckedIOException e) {
           throw e.getCause();
      }
  

Checked Exceptions

• Apache commons.lang FailableStream
• sneakyfun, Lombok @SneakyThrow

  sneaky(targetLink -> {
      Files.createDirectories(targetLink.getParent());
      Files.copy(file, targetLink, StandardCopyOption.REPLACE_EXISTING, LinkOption.NOFOLLOW_LINKS);
  });
  

• Custom gatherer (for map/filter)

  interface CheckedFunction<T, R> { R apply(T arg) throws Exception; }
  
  <T, R> Gatherer<T, ?, R> checkedMap(CheckedFunction<? super T,? extends R> mapper) {
      return Gatherer.<T, R>of(
          Gatherer.Integrator.ofGreedy((_, element, downstream) -> {
              try {
                  return downstream.push(mapper.apply(element));
              } catch (Exception ex) {
                  throw new RuntimeException("checkedMap", ex);
              }
          }));
  }
  

• or collector (for forEach):

  walk.collect(checkedForEach(file -> ...))
  

Benchmarking

• JMH benchmark with pre-initialized state.randomPointArray, comparing

  List<Point> points = List.of(state.randomPointArray);
  Point largest = points.get(0);
  for (int i = 1; i < points.size(); i++) {
      Point p = points.get(i);
      if (p.x % 2 == 0 && p.y % 2 == 0) {
          if (p.x * p.x + p.y * p.y > largest.x * largest.x + largest.y * largest.y)
              largest = p;
      }
  }
  

  with

  Point largest = Stream.of(state.randomPointArray)
      .filter(p -> p.x % 2 == 0 && p.y % 2 == 0)
      .max(Comparator.comparingInt(p -> p.x * p.x + p.y * p.y))
      .orElseThrow();
  

• Streams are twice as fast!

  MyBenchmark.withCollection  avgt  100  206.1 ± 8.9 ms/op
  MyBenchmark.withStream      avgt  100   89.3 ± 0.6 ms/op
  

• Did you know that List.of makes a copy?
• With pre-initialized list:

  MyBenchmark.withCollection  avgt  100  199.2 ± 8.9 ms/op
  MyBenchmark.withStream      avgt  100  212.6 ± 8.7 ms/op
  

• Benchmarks by Nikolai Parlog indicate modest stream overhead

Memory Consumption

• Stream pipeline = data structure for managing stream execution
  • A node for every intermediate operation
  • Inner class instances for ops
  • Spliterators
• Structure set up once
• When the JIT kicks in, no per-iteration allocations
• Not usually a concern for iterating over large data
• More noticeable with stream usage inside tight loop
• Example: Looping 100_000 times over stream calculation produced about 20MB of garbage
• Observe with JFR, JOL
• Could be an issue with nested streams, or window gatherers

Concurrency

• Parallel stream splits into n subranges (n = number of cores)

  Point largest = Stream.of(state.randomPointArray)
      .parallel()
      .filter(p -> p.x % 2 == 0 && p.y % 2 == 0)
      .max(Comparator.comparingLong(p -> (long) p.x * p.x + (long) p.y * p.y))
      .orElseThrow();
  

• Subrange tasks execute on global fork-join pool
• Results are assembled in order
• Happy day scenario: n-fold speedup

  ConcurrentBenchmark.withParallelStream  avgt  100   19.082 ± 0.212  ms/op
  ConcurrentBenchmark.withSerialStream    avgt  100  203.989 ± 3.187  ms/op
  

• Not all days are happy

  long largest = Stream.of(state.randomPointArray)
      .parallel()
      .mapToLong(p -> (long) p.x * p.x + (long) p.y * p.y)
      .filter(p -> p > 0.5)
      .limit(ITERATIONS / 10)
      .max()
      .orElseThrow();
  

  ConcurrentBenchmark2.withParallelStream  avgt  100  27.127 ± 1.876  ms/op
  ConcurrentBenchmark2.withSerialStream    avgt  100  10.579 ± 0.339  ms/op
  

Blocking Tasks

• Parallel streams do not work well for tasks that mostly block

  String task(int i) {
      try {
          Thread.sleep(1000);
      } catch (InterruptedException _) {
      }
      return i + " " + Thread.currentThread().toString();
  }
  

• Sequential stream: 60 seconds:

  IntStream.range(0, 60).mapToObj(this::task).toList()
  

• Parallel stream: 5 seconds (on my machine)

  IntStream.range(0, 60).parallel().mapToObj(this::task).toList()
  

• But it blocks the global fork-join pool
• Really bad if other concurrent tasks do the same
• Isn’t that what virtual threads are for?
• mapConcurrent: 1 second

  IntStream.range(0, 60).boxed().gather(Gatherers.mapConcurrent(60, this::task)).toList()
  

• Also check out Grzegorz Piwowarek’s Parallel Collectors

Case Study: Phone Mnemonics

• Given a phone number, what word or word sequence spells it on a touch-tone pad?
• For example, 435 569 6753 → "HELLO WORLD" (and maybe others)
• From Martin Odersky's 2011 State of Scala, adapted from this paper by Lutz Prechelt
• Solution follows my 2015 Scala course
• 68 lines of code in JTaccuino notebook
• That's 75% less than the median for Java in Prechelt's paper
• For better or worse, no loops, no recurson. Only streams (and a custom gatherer)
• The tempation to nest streams is irresistible
• No fun: Fixing compiler errors, debugging nested streams
• Pretty opaque even after cleanup
• Would need another round of refactoring
• Verdict: Until we all get more fluent with flatMap and fold, budget time for refactoring and documentation

The End

• Favor streams when they are clearer than loops
• Good use cases:
  • Max, min, sorting
  • sums and averages
  • finding, counting, collecting matches
• Favor method expressions
• Refactor complex lambdas, nested streams into helper methods
• “Functional” is not pixie dust—some loops should not be streams
• Embrace common idioms (collecting into maps, skipping optionals)
• Don’t fret over serial performance
• Understand the pipeline for parallel performance
• Questions?

Questions?