ここ最近 Reactor をいじって一定間隔の無限ストリームを生成するなどしていたこともあって、まとめることにした。タイトルに数字が入っていると読まれやすいっぽいので、タイトルには「3つのパターン」と書いたが、実際に flatMap 、 flatMapSequential 、 concatMap の3つなので、あながち間違っているわけではない。
Flux 導入
Flux は 0 以上のデータを持つデータストリーム。例えば 0.25 秒おきに 0 から 1 ずつ増えるデータを記述すると次のようになる。
@Grab('io.projectreactor:reactor-core:3.1.8.RELEASE') import reactor.core.publisher.* def flux = Flux.interval(Duration.ofMillis(250)).take(4)
では、この Flux<Long> の数値を文字列として表示したいので、 String になるように map する。
def flux = Flux.interval(Duration.ofMillis(250)).take(4) flux.map { i -> "test $i" }
次に、時刻を表示したい。そこで、この String と現在の時刻を文字列に埋め込んで表示する。
def flux = Flux.interval(Duration.ofMillis(250)).take(4) flux.map { i -> "test $i" } .map { s -> "[${LocalDate.now()}] $s" } .subscribe { logger.info(it) }
Flux の中の処理はメインスレッドで subscribe を呼び出してから開始されるので、メインスレッドよりも遅れて実行される。そのため先に終わったメインスレッドが Flux の中の処理を待ち合わせるために、 CountDownLatch などの同期化するための機構を埋め込まないとならない。 CountDownLatch を Flux の処理が終わったあとに countDown されるように onTerminate に処理を加えるなどすると、次のようなプログラムになる。
@Grab('io.projectreactor:reactor-core:3.1.8.RELEASE') @Grab('org.slf4j:slf4j-simple:1.7.25') import java.util.concurrent.* import java.time.* import java.time.format.* import reactor.core.publisher.* import reactor.core.scheduler.* import org.slf4j.* def logger = LoggerFactory.getLogger('groovy') Flux<Long> flux = Flux.interval(Duration.ofMillis(250)).take(4) def latch = new CountDownLatch(1) def disp = flux .map { i -> "test $i" } .map { s -> "[${LocalTime.now()}] $s" } .doOnTerminate { latch.countDown() } .subscribe { logger.info(it) } latch.await() disp.dispose()
これを実行すると、次のように表示される。
[parallel-6] INFO groovy - [02:00:41.573] test 0 [parallel-6] INFO groovy - [02:00:41.822] test 1 [parallel-6] INFO groovy - [02:00:42.075] test 2 [parallel-6] INFO groovy - [02:00:42.322] test 3
これでまずは Flux の導入ができた。
flatten する
以下では上で作った Flux を起点に一定間隔でデータを投げる Flux を作っていく。各 Flux の起点から 400 ms 感覚でエミットされるデータを3つずつ取るものとする。次の表はデータがエミットされる時刻の一覧(単位は ms)
| 1st | 2nd-0 | 2nd-1 | 2nd2 |
|---|---|---|---|
| 0 | 0 |
400 |
800 |
| 1 | 250 |
650 |
1050 |
| 2 | 500 |
900 |
1300 |
| 3 | 750 |
1150 |
1550 |
次のようなクラスを導入して、どのデータがいつエミットされたのかを記録する
class Sample { final long first final long second final LocalTime created Sample(first, second) { this.first = first this.second = second this.created = LocalTime.now() } @Override String toString() { "($created) $first - $second" } }
また、次の関数によって long から Sample を作る
def longToSample = {long fst -> Flux.interval(Duration.ofMillis(400)) .take(3) .map { long snd -> new Sample(fst, snd) } }
発生した時刻をキーにマージする flatMap
flatMap は発生した時刻をキーにしてマージしていく。したがって、想定されるデータ列は先程の表から、 0-0(0 ms) -> 1-0(250 ms) -> 0-1(400 ms) -> 2-0(500 ms) -> 1-1(650 ms) -> 3-0(750 ms) -> 0-2(800 ms)... となることが予想される。
Flux<Long> flux = Flux.interval(Duration.ofMillis(250)).take(4) def disp = flux .flatMap(longToSample) .map { s -> "[${LocalTime.now()}] $s" } .doOnTerminate { latch.countDown() } .subscribe { logger.info(it) } latch.await() disp.dispose()
この場合、出力は次のようになる
[parallel-5] INFO groovy - [10:32:07.413] (10:32:07.413) 0 - 0 [parallel-6] INFO groovy - [10:32:07.647] (10:32:07.647) 1 - 0 [parallel-5] INFO groovy - [10:32:07.798] (10:32:07.798) 0 - 1 [parallel-7] INFO groovy - [10:32:07.900] (10:32:07.900) 2 - 0 [parallel-6] INFO groovy - [10:32:08.047] (10:32:08.046) 1 - 1 [parallel-8] INFO groovy - [10:32:08.151] (10:32:08.151) 3 - 0 [parallel-5] INFO groovy - [10:32:08.199] (10:32:08.199) 0 - 2 [parallel-7] INFO groovy - [10:32:08.297] (10:32:08.297) 2 - 1 [parallel-6] INFO groovy - [10:32:08.446] (10:32:08.446) 1 - 2 [parallel-8] INFO groovy - [10:32:08.551] (10:32:08.551) 3 - 1 [parallel-7] INFO groovy - [10:32:08.696] (10:32:08.696) 2 - 2 [parallel-8] INFO groovy - [10:32:08.948] (10:32:08.948) 3 - 2
複数の系列で並行になんらかの処理を行いたい場合に用いるとよさそうである
元の順序をキープして新たな間隔でエミットする concatMap
concatMap でまとめられた Flux は元の順序(1st)を保持しつつ、新しく指定された間隔でエミットされる。想定される順番は 0-0 -> 0-1 -> 0-2 -> 1-0 -> 1-1 -> ... となり、それらの間隔が 400 ms となる(最初の 250 ms は無視される)。
Flux<Long> flux = Flux.interval(Duration.ofMillis(250)).take(4) def disp = flux .concatMap(longToSample) .map { s -> "[${LocalTime.now()}] $s" } .doOnTerminate { latch.countDown() } .subscribe { logger.info(it) } latch.await() disp.dispose()
この場合の出力は次のようになる
[parallel-3] INFO groovy - [10:36:56.553] (10:36:56.553) 0 - 0 [parallel-3] INFO groovy - [10:36:56.939] (10:36:56.939) 0 - 1 [parallel-3] INFO groovy - [10:36:57.342] (10:36:57.342) 0 - 2 [parallel-4] INFO groovy - [10:36:57.743] (10:36:57.743) 1 - 0 [parallel-4] INFO groovy - [10:36:58.145] (10:36:58.145) 1 - 1 [parallel-4] INFO groovy - [10:36:58.543] (10:36:58.543) 1 - 2 [parallel-5] INFO groovy - [10:36:58.948] (10:36:58.948) 2 - 0 [parallel-5] INFO groovy - [10:36:59.346] (10:36:59.346) 2 - 1 [parallel-5] INFO groovy - [10:36:59.748] (10:36:59.748) 2 - 2 [parallel-6] INFO groovy - [10:37:00.153] (10:37:00.153) 3 - 0 [parallel-6] INFO groovy - [10:37:00.552] (10:37:00.552) 3 - 1 [parallel-6] INFO groovy - [10:37:00.949] (10:37:00.949) 3 - 2
元の順序をキープ(concatMap)、各要素は指定時刻に生成(flatMap)される flatMapSequential
flatMapSequential は concatMap と flatMap を組み合わせたようなもので、 要素の順序は元の順序(1st -> 2nd の順序)をキープしつつ、データのエミットタイミングもキープされる動作をとる。なお元の順序を飛び越してしまったデータはバッファリングされる。
def disp = flux .flatMapSequential(longToSample) .map { s -> "[${LocalTime.now()}] $s" } .doOnTerminate { latch.countDown() } .subscribe { logger.info(it) } latch.await() disp.dispose()
出力は次のようになる
[parallel-8] INFO groovy - [12:06:06.070] (12:06:06.069) 0 - 0 [parallel-8] INFO groovy - [12:06:06.455] (12:06:06.455) 0 - 1 [parallel-8] INFO groovy - [12:06:06.855] (12:06:06.855) 0 - 2 [parallel-8] INFO groovy - [12:06:06.856] (12:06:06.304) 1 - 0 [parallel-8] INFO groovy - [12:06:06.856] (12:06:06.702) 1 - 1 [parallel-9] INFO groovy - [12:06:07.104] (12:06:07.104) 1 - 2 [parallel-9] INFO groovy - [12:06:07.105] (12:06:06.551) 2 - 0 [parallel-9] INFO groovy - [12:06:07.105] (12:06:06.953) 2 - 1 [parallel-10] INFO groovy - [12:06:07.350] (12:06:07.350) 2 - 2 [parallel-10] INFO groovy - [12:06:07.351] (12:06:06.805) 3 - 0 [parallel-10] INFO groovy - [12:06:07.351] (12:06:07.205) 3 - 1 [parallel-11] INFO groovy - [12:06:07.605] (12:06:07.605) 3 - 2
データ 1-0 は データ 0-1 よりも先にエミットされるが、バッファリングされて 0-2 の後に出てきているのがわかる
おわり
