Слияние потоков в Java

1. Обзор

В этой быстрой статье мы объясним различные способы слияния потоков Java , что не очень интуитивно понятно.

2. Использование простой Java

Класс JDK 8 Stream имеет несколько полезных статических служебных методов. Давайте подробнее рассмотрим метод concat() .

2.1. Слияние двух потоков

Самый простой способ объединить 2 Stream — использовать статический метод Stream.concat() :

@Test
public void whenMergingStreams_thenResultStreamContainsElementsFromBoth() {
    Stream<Integer> stream1 = Stream.of(1, 3, 5);
    Stream<Integer> stream2 = Stream.of(2, 4, 6);

    Stream<Integer> resultingStream = Stream.concat(stream1, stream2);

    assertEquals(
      Arrays.asList(1, 3, 5, 2, 4, 6),
      resultingStream.collect(Collectors.toList()));
}

2.2. Объединение нескольких потоков

Когда нам нужно объединить более двух потоков, все становится немного сложнее. Одна из возможностей состоит в том, чтобы объединить первые два потока, затем объединить результат со следующим и так далее.

Следующий фрагмент кода показывает это в действии:

@Test
public void given3Streams_whenMerged_thenResultStreamContainsAllElements() {
    Stream<Integer> stream1 = Stream.of(1, 3, 5);
    Stream<Integer> stream2 = Stream.of(2, 4, 6);
    Stream<Integer> stream3 = Stream.of(18, 15, 36);

    Stream<Integer> resultingStream = Stream.concat(
      Stream.concat(stream1, stream2), stream3);

    assertEquals(
      Arrays.asList(1, 3, 5, 2, 4, 6, 18, 15, 36),
      resultingStream.collect(Collectors.toList()));
}

Как мы видим, этот подход становится невозможным для большего количества потоков. Конечно, мы можем создать промежуточные переменные или вспомогательные методы, чтобы сделать его более читабельным, но вот вариант получше:

@Test
public void given4Streams_whenMerged_thenResultStreamContainsAllElements() {
    Stream<Integer> stream1 = Stream.of(1, 3, 5);
    Stream<Integer> stream2 = Stream.of(2, 4, 6);
    Stream<Integer> stream3 = Stream.of(18, 15, 36);
    Stream<Integer> stream4 = Stream.of(99);

    Stream<Integer> resultingStream = Stream.of(
      stream1, stream2, stream3, stream4)
      .flatMap(i -> i);

    assertEquals(
      Arrays.asList(1, 3, 5, 2, 4, 6, 18, 15, 36, 99),
      resultingStream.collect(Collectors.toList()));
}

Здесь происходит следующее:

• Сначала мы создаем новый поток , содержащий 4 потока , в результате чего получается Stream<Stream<Integer>> .
• Затем мы flatMap() преобразуем это в Stream<Integer> , используя функцию идентификации.

3. Использование StreamEx

StreamEx — это библиотека Java с открытым исходным кодом, которая расширяет возможности Java 8 Streams. Он использует класс StreamEx в качестве расширения интерфейса JDK Stream .

3.1. Объединение потоков _

Библиотека StreamEx позволяет нам объединять потоки, используя метод экземпляра append() :

@Test
public void given4Streams_whenMerged_thenResultStreamContainsAllElements() {
    Stream<Integer> stream1 = Stream.of(1, 3, 5);
    Stream<Integer> stream2 = Stream.of(2, 4, 6);
    Stream<Integer> stream3 = Stream.of(18, 15, 36);
    Stream<Integer> stream4 = Stream.of(99);

    Stream<Integer> resultingStream = StreamEx.of(stream1)
      .append(stream2)
      .append(stream3)
      .append(stream4);

    assertEquals(
      Arrays.asList(1, 3, 5, 2, 4, 6, 18, 15, 36, 99),
      resultingStream.collect(Collectors.toList()));
}

Поскольку это метод экземпляра, мы можем легко связать его и добавить несколько потоков.

Обратите внимание, что мы также можем создать список из потока с помощью toList() , если мы приведем результирующую переменную Stream к типу StreamEx .

3.2. Слияние потоков с помощью prepend()

StreamEx также содержит метод prepend() , который добавляет элементы один перед другим :

@Test
public void given3Streams_whenPrepended_thenResultStreamContainsAllElements() {
    Stream<String> stream1 = Stream.of("foo", "bar");
    Stream<String> openingBracketStream = Stream.of("[");
    Stream<String> closingBracketStream = Stream.of("]");

    Stream<String> resultingStream = StreamEx.of(stream1)
      .append(closingBracketStream)
      .prepend(openingBracketStream);

    assertEquals(
      Arrays.asList("[", "foo", "bar", "]"),
      resultingStream.collect(Collectors.toList()));
}

4. Использование Jooλ

jOOλ — это библиотека, совместимая с JDK 8, которая предоставляет полезные расширения для JDK. Самая важная абстракция потока здесь называется Seq . Обратите внимание, что это последовательный и упорядоченный поток, поэтому вызов parallel() не будет иметь никакого эффекта.

4.1. Объединение потоков

Как и в библиотеке StreamEx, в jOOλ есть метод append() :

@Test
public void given2Streams_whenMerged_thenResultStreamContainsAllElements() {
    Stream<Integer> seq1 = Stream.of(1, 3, 5);
    Stream<Integer> seq2 = Stream.of(2, 4, 6);

    Stream<Integer> resultingSeq = Seq.ofType(seq1, Integer.class)
      .append(seq2);

    assertEquals(
      Arrays.asList(1, 3, 5, 2, 4, 6),
      resultingSeq.collect(Collectors.toList()));
}

Также есть удобный метод toList() , если мы приводим результирующую переменную Seq к типу jOOλ Seq .

4.2. Слияние потоков с prepend()

Как и ожидалось, поскольку существует метод append() , в jOOλ есть и метод prepend() :

@Test
public void given3Streams_whenPrepending_thenResultStreamContainsAllElements() {
    Stream<String> seq = Stream.of("foo", "bar");
    Stream<String> openingBracketSeq = Stream.of("[");
    Stream<String> closingBracketSeq = Stream.of("]");

    Stream<String> resultingStream = Seq.ofType(seq, String.class)
      .append(closingBracketSeq)
      .prepend(openingBracketSeq);

    Assert.assertEquals(
      Arrays.asList("[", "foo", "bar", "]"),
      resultingStream.collect(Collectors.toList()));
}

5. Вывод

Мы видели, что слияние потоков с помощью JDK 8 относительно простое. Когда нам нужно выполнить много слияний, может быть полезно использовать библиотеку StreamEx или jOOλ для удобочитаемости.

Вы можете найти исходный код на GitHub .