並列処理の種類

ZIOには並列処理の仕組みが組み込まれており、手軽に通常の処理を並列処理にすることが出来ます。
ZIOの並列処理は一般的なものは以下の2種類に大別されます。

1. ZIO.forkによるもの
2. 関数名の後ろにParが付くもの

これらについて順に見ていきます。

ZIO.forkによる並列処理

以下のようにコードを記載します。

object MainApp extends ZIOAppDefault {
  def run: ZIO[Any, Throwable, Unit] = ApplicationService.consoleOutput().provide(ZLayer.fromZIO(ZIO.attempt {
    import java.text.SimpleDateFormat
    // 任意の日付文字列
    val inpDateStr = "2023/07/25 17:46:00"

    // 取り扱う日付の形にフォーマット設定
    val sdformat = new SimpleDateFormat("yyyy/MM/dd HH:mm:ss")

    // Date型に変換( DateFromatクラスのparse() )
    sdformat.parse(inpDateStr)
  }), ApplicationServiceImpl.layer)
}

case class ApplicationServiceImpl(currentDate: Date) extends ApplicationService {
  override def consoleOutput(): ZIO[Any, Throwable, Unit] = for {
    fiber <- (for {
      _ <- ZIO.attempt(Thread.sleep(3000))
      _ <- Console.printLine("並列処理実行")
      result <- ZIO.succeed("並列処理:成功")
    } yield result).fork
    _ <- Console.printLine(
      s"${new SimpleDateFormat("yyyy/MM/dd HH:mm:ss").format(currentDate)} Hello, World!"
    )
    result <- fiber.join
    _ <- Console.printLine(result)
  } yield ()
}

このプログラムを実行すると以下の出力がコンソールから得られると思います。

2023/07/25 17:46:00 Hello, World!
並列処理実行
並列処理:成功

この結果から日時を出力する処理が先に実行され、後から先に記載されている部分が実行され、並列処理が実行されていることが確認できたと思います。
このプログラムについて詳しく見ていきましょう。

ZIO.fork

まず最初に以下の部分のようにfork関数を呼び出します。

    fiber <- (for {
      _ <- ZIO.attempt(Thread.sleep(3000))
      _ <- Console.printLine("並列処理実行")
      result <- ZIO.succeed("並列処理:成功")
    } yield result).fork

これによりfor yield句内の処理が並列して実行されるように指定されます。
返り値を受け取ったfiber変数はString型ではないため、このままでは後続の処理に使えません。
後続の処理に使うためには続いてjoin関数を呼び出します。

join関数

並列処理されている処理の返り値を受け取るためには最終的にjoin関数を呼び出す必要があります。

result <- fiber.join

このresult変数の型はString型であり、他の処理にも利用できる型です。
「並列処理:成功」という文字列であり、コンソールへの出力に使っています。

以上のようにZIO型に対してfork関数とjoin関数をセットで使うことにより簡単に並列処理を実行することができます。

後ろにParが付く関数による並列処理

関数名の後ろにParが付くものを試す前にParが付かない場合の挙動を確認します。
以下の様にコードを修正します。

case class ApplicationServiceImpl(currentDate: Date) extends ApplicationService {
  override def consoleOutput(): ZIO[Any, Throwable, Unit] = for {
    // 以下に関して警告が表示されるかもしれませんが、動作確認が主目的のため一旦無視します。
    _ <- ZIO.foreach(List.range(1, 100))(n => Console.printLine(n))
  } yield ()
}

ZIO.foreach関数はListと関数を引数に取り、リストの各要素に対して引数の関数を適用し、最終的にZIO型のListを返す関数になります。
修正後実行すると以下の様な出力が得られると思います。

1
2
3
4
5
(省略)
95
96
97
98
99

1から99までの数が何度実行し返したとしても順に出力されるはずです。
これは処理が並列ではなく直列に実行されていることを意味しています。

次に以下の様にZIO.foreach関数をZIO.foreachPar関数に修正します。

    _ <- ZIO.foreachPar(List.range(1, 100))(n => Console.printLine(n))

これを実行すると以下の様な出力が得られます。

36
86
3
12
14
79
42
(省略)
24
25
26
21
23
22

先ほどとは打って変わり、順序関係なく数字が出力されます。
また、繰り返す毎に異なる出力結果になるはずです。
このことからZIO.foreachPar関数により並列実行が行われたことがわかります。
ZIO.foreachPar関数の他にも同様のZIO.partitionPar関数やZIO.mergeAllPar関数があります。

Fiberについて(並列処理の仕組み)

最後にこれらの並列処理の仕組みについて簡単に触れたいと思います。
このようなZIOの並列処理は一般的なJavaプログラムの並列処理で使われるThreadではなくFiberによって行われます。

FiberについてはThreadを改良したものとなっており、1つのThreadに対して複数のFiberが対応する関係になっています。
以下の様な長所があります。

1. Threadと比べて生成可能な数が非常に多い
2. 生成コストがThreadに比べて低い
3. Fiberは処理に失敗した場合と成功した場合の返り値について型付けされている

以下のページにFiberの概要が載っています。

終わりに

今回はZIOの並列処理について見てきました。
今回でバッチ編は終わりです。
次回からバックエンド編に入ります。
最初はzio-httpによるGETメソッド、POSTメソッドの処理等について見ていきます。

前章:ZIOライブラリ利用(DB接続)
次章:zio-http

演習

1. 今回の2つの並列処理について実際に動かして挙動を確かめてください。
   1. ZIO.fork関数を使ったものの解答例は以下になります。
      https://github.com/hatuda/zio-practice/tree/CHAPTER5-Fork
   2. 末尾がParとなっている関数を使ったものの解答例は以下になります。
      https://github.com/hatuda/zio-practice/tree/CHAPTER5-Par