1概要

この記事では、https://projectreactor.io/[プロジェクトの反応器]で

Publishers

を組み合わせるさまざまな方法について説明します。


2 Mavenの依存関係


https://search.maven.org/classic/#search%7Cga%7C1%7Cg%3A%22io.projectreactor%22%20AND%20a%3A%22reactor-core%22

[Project Reactorを使って例を設定しましょう依存関係: 

<dependency>
    <groupId>io.projectreactor</groupId>
    <artifactId>reactor-core</artifactId>
    <version>3.1.4.RELEASE</version>
</dependency>
<dependency>
    <groupId>io.projectreactor</groupId>
    <artifactId>reactor-test</artifactId>
    <version>3.1.4.RELEASE</version>
    <scope>test</scope>
</dependency>


3出版社の結合


Flux <T>

または__Mono <T>を操作する必要があるシナリオでは、ストリームを組み合わせる方法はいくつかあります。


Flux <T>

クラスの静的メソッド（

concat、concatWith、merge、zip

、

combineLatest.

など）の使い方を説明するための例をいくつか作成しましょう。

この例では

Flux <Integer>

という2つの発行元、つまり

EvenNumbers

を使用します。これは

Integer

の

Flux

であり、1（

min

variable）から始まり5（

max

variable）までに制限された偶数のシーケンスを持ちます。


oddNumbers

、また奇数の

Integer

型の

Flux

を作成します。 

Flux<Integer> evenNumbers = Flux
  .range(min, max)
  .filter(x -> x % 2 == 0);//i.e. 2, 4

Flux<Integer> oddNumbers = Flux
  .range(min, max)
  .filter(x -> x % 2 > 0); //ie. 1, 3, 5


3.1. c


_oncat（）

_



concat

メソッドは入力の連結を実行し、ソースから発行された要素を下流に転送します。

連結は、最初のソースを順次サブスクライブし、次にそれが完了するのを待ってから次のソースをサブスクライブするという具合に、最後のソースが完了するまで同様に続きます。エラーが発生するとすぐにシーケンスが中断され、ダウンストリームに転送されます。

これは簡単な例です。 

@Test
public void givenFluxes__whenConcatIsInvoked__thenConcat() {
    Flux<Integer> fluxOfIntegers = Flux.concat(
      evenNumbers,
      oddNumbers);

    StepVerifier.create(fluxOfIntegers)
      .expectNext(2)
      .expectNext(4)
      .expectNext(1)
      .expectNext(3)
      .expectNext(5)
      .expectComplete()
      .verify();
}


3.2.



concatWith（）


結果として静的メソッド

concatWith

を使用して、タイプ

Flux <T>

の2つのソースの連結を生成します。 

@Test
public void givenFluxes__whenConcatWithIsInvoked__thenConcatWith() {
    Flux<Integer> fluxOfIntegers = evenNumbers.concatWith(oddNumbers);

   //same stepVerifier as in the concat example above
}


3.3. c


_ombineLatest（）

_


Flux静的メソッド

combineLatest

は、各パブリッシャソースから最後に公開された値の組み合わせによって提供されるデータを生成します。

これは、2つの

Publisher

ソースと1つの

BiFunction

をパラメーターとして使用したこのメソッドの使用例です。 

@Test
public void givenFluxes__whenCombineLatestIsInvoked__thenCombineLatest() {
    Flux<Integer> fluxOfIntegers = Flux.combineLatest(
      evenNumbers,
      oddNumbers,
      (a, b) -> a + b);

    StepVerifier.create(fluxOfIntegers)
      .expectNext(5)//4 + 1
      .expectNext(7)//4 + 3
      .expectNext(9)//4 + 5
      .expectComplete()
      .verify();
}

ここでは、関数

combineLatest

が、最新の

evenNumbers

（

4

）の要素と

oddNumbers（1,3,5）

の要素を使用して関数 “a b”を適用して、シーケンス

5,7,9

が生成されることがわかります。


3.4.

マージ（）



merge

関数は、配列に含まれている

Publisher

シーケンスのデータを、インターリーブされたマージ済みシーケンスにマージします。 

@Test
public void givenFluxes__whenMergeIsInvoked__thenMerge() {
    Flux<Integer> fluxOfIntegers = Flux.merge(
      evenNumbers,
      oddNumbers);

    StepVerifier.create(fluxOfIntegers)
      .expectNext(2)
      .expectNext(4)
      .expectNext(1)
      .expectNext(3)
      .expectNext(5)
      .expectComplete()
      .verify();
}

注意すべき興味深いことは、

__concat（


lazy subscription


）

__とは対照的に、ソースは熱心に購読されているということです。

ここで、Publishersの要素間に遅延を挿入すると、

merge

関数の結果が異なることがわかります。 

@Test
public void givenFluxes__whenMergeWithDelayedElementsIsInvoked__thenMergeWithDelayedElements() {
    Flux<Integer> fluxOfIntegers = Flux.merge(
      evenNumbers.delayElements(Duration.ofMillis(500L)),
      oddNumbers.delayElements(Duration.ofMillis(300L)));

    StepVerifier.create(fluxOfIntegers)
      .expectNext(1)
      .expectNext(2)
      .expectNext(3)
      .expectNext(5)
      .expectNext(4)
      .expectComplete()
      .verify();
}

3.5.


mergeSequential（）



mergeSequential

メソッドは、配列で提供された

Publisher

シーケンスのデータを順序付けられたマージシーケンスにマージします。


concat

とは異なり、情報源は熱心に購読しています。

また、

merge

とは異なり、それらの発行された値は、サブスクリプション順に最終シーケンスにマージされます。 

@Test
public void testMergeSequential() {
    Flux<Integer> fluxOfIntegers = Flux.mergeSequential(
      evenNumbers,
      oddNumbers);

    StepVerifier.create(fluxOfIntegers)
      .expectNext(2)
      .expectNext(4)
      .expectNext(1)
      .expectNext(3)
      .expectNext(5)
      .expectComplete()
      .verify();
}

3.6


mergeDelayError（）



mergeDelayError

は、配列に含まれているパブリッシャシーケンスのデータをインターリーブされたマージシーケンスにマージします。


concat

とは異なり、情報源は熱心に購読しています。

この静的な

merge

メソッドの変形は、残りのマージバックログが処理されるまでエラーを遅らせます。

これは__mergeDelayErrorの例です。 

@Test
public void givenFluxes__whenMergeWithDelayedElementsIsInvoked__thenMergeWithDelayedElements() {
    Flux<Integer> fluxOfIntegers = Flux.mergeDelayError(1,
      evenNumbers.delayElements(Duration.ofMillis(500L)),
      oddNumbers.delayElements(Duration.ofMillis(300L)));

    StepVerifier.create(fluxOfIntegers)
      .expectNext(1)
      .expectNext(2)
      .expectNext(3)
      .expectNext(5)
      .expectNext(4)
      .expectComplete()
      .verify();
}



3.7. と合併（）


静的メソッド

mergeWith

は、この

Flux

と

Publisher

からのデータをインターリーブされたマージ済みシーケンスにマージします。

繰り返しになりますが、

concat

とは異なり、内部の情報源は熱心に購読しています。 

@Test
public void givenFluxes__whenMergeWithIsInvoked__thenMergeWith() {
    Flux<Integer> fluxOfIntegers = evenNumbers.mergeWith(oddNumbers);

   //same StepVerifier as in "3.4. Merge"
    StepVerifier.create(fluxOfIntegers)
      .expectNext(2)
      .expectNext(4)
      .expectNext(1)
      .expectNext(3)
      .expectNext(5)
      .expectComplete()
      .verify();
}


3.8. z


_ip（）

_


静的メソッド

zip

は、複数のソースを集約します。つまり、すべてのソースが1つの要素を発行するのを待って、これらの要素を（提供された組み合わせ関数によって構築された）出力値に組み合わせます。

オペレータは、いずれかのソースが完了するまでそうし続けます。 

@Test
public void givenFluxes__whenZipIsInvoked__thenZip() {
    Flux<Integer> fluxOfIntegers = Flux.zip(
      evenNumbers,
      oddNumbers,
      (a, b) -> a + b);

    StepVerifier.create(fluxOfIntegers)
      .expectNext(3)//2 + 1
      .expectNext(7)//4 + 3
      .expectComplete()
      .verify();
}


evenNumbers

からペアリングする要素が残っていないため、

oddNumbers

Publisherからの要素5は無視されます。


3.9. z


_ipWith（）

_



zipWith

は

zip

と同じメソッドを実行しますが、2人のパブリッシャでのみ実行されます。 

@Test
public void givenFluxes__whenZipWithIsInvoked__thenZipWith() {
    Flux<Integer> fluxOfIntegers = evenNumbers
     .zipWith(oddNumbers, (a, b) -> a **  b);

    StepVerifier.create(fluxOfIntegers)
      .expectNext(2) //2 **  1
      .expectNext(12)//4 **  3
      .expectComplete()
      .verify();
}


4結論

このクイックチュートリアルでは、

Publishers.

を組み合わせる複数の方法を発見しました。

いつものように、例はhttps://github.com/eugenp/tutorials/tree/master/reactor-core[GitHubの上に]で利用可能です。