JavaのParallelStreamでは、Transactionが別々になる

こんにちは。エキサイト株式会社の三浦です。

今まで何回か、ParallelStreamを使ったJavaの並列処理に関する記事を上げてきました。

今回は、ParallelStreamとDBのTransactionを一緒に使う際の注意点を説明します。

JavaのParallelStream
DBのTransaction
ParallelStreamとTransaction
最後に

JavaのParallelStream

以前、以下の記事で、JavaではParallelStreamを使うことで並列処理を行うことができると説明しました。

tech.excite.co.jp

非常にシンプルに並列処理を行うことができるのですが、いくつか逐次処理の際にはなかった注意点があります。

その一つが、DBのTransaction処理です。

DBのTransaction

DBのTransactionとは、端的に言うと、DBに対する動作に不整合が起きないことを保証する仕組みです。

例えば、あるDBが記事情報を管理しているとして、DBに記事データを入れるときに以下の流れになっているとしましょう。

DBに記事のタイトルデータを入れる
DBに記事の本文データを入れる
DBに記事の公開日データを入れる

public void createArticle() {
    this.setTitle("タイトル");
    this.setBody("本文");
    this.setPublishDate(LocalDateTime.now());
}

特に問題なく以上の操作ができればいいですが、何かしらの事情で途中でエラーが発生することも考えられます。

Transactionを使わない場合、その事も考えて、例えば「記事のタイトルは存在するが本文が存在しない場合はエラーデータ」などを考慮する必要があるかもしれませんが、Transactionを使うともっとシンプルに考えることができるようになります。

Transactionを使う場合、今回は以下のような流れになります。

Transaction開始
DBに記事のタイトルデータを入れる
DBに記事の本文データを入れる
DBに記事の公開日データを入れる
Transaction終了

@Transactional // このアノテーションにより、メソッド内の処理にトランザクションがかけられます
public void createArticle() {
    this.setTitle("タイトル");
    this.setBody("本文");
    this.setPublishDate(LocalDateTime.now());
}

Transactionの開始・終了の手間は増えてしまいますが、これをすることによって、途中でエラーが起きた場合、それまでDBに対して行っていた処理をすべてなかったことにしてくれます。

例えば、「DBに記事のタイトルデータを入れる」までは成功し、「DBに記事の本文データを入れる」のタイミングでエラーが起きてしまった場合、Transactionが「DBに記事のタイトルデータを入れる」という処理をなかったことにしてくれるので、「記事のタイトルは存在するが本文が存在しない」という中途半端な状態は発生せず、常に「すべて成功していたら完全な記事データが存在し、どこかで失敗していたらそもそも記事データが存在しない」ことだけを考えれば済むようになります。

DBへの書き込み処理をする場合は、ほとんどの場合Transactionは必須と言っても良いでしょう。

ParallelStreamとTransaction

ですが、ParallelStreamとTransactionを併用する場合は注意が必要です。

例えば、以下の状況を考えてみましょう。

Transaction開始
並列処理で、以下の処理を行う
1. DBに記事のタイトルデータを入れる
2. DBに記事の本文データを入れる
3. DBに記事の公開日データを入れる
すべての並列処理が終了したらTransaction終了

@Transactional
public void createArticleListInParallel() {
    final List<Article> articleList = List.of(
        new Article("タイトル", "本文", LocalDateTime.now()),
        // ...
    );

    articleList
            .parallelStream()
            .forEach(article -> this.createArticle(article));
}

private void createArticle(Article article) {
    this.setTitle(article.getTitle());
    this.setBody(article.getBody());
    this.setPublishDate(article.getPublishDate());
}

先程のTransactionを考えれば、すべての並列処理でTransactionが効くように考えてしまうかもしれませんが、実は並列処理の中ではTransactionは効いていません。

そのため、ある並列処理の途中（例えば本文データを入れる処理）でエラーが起きた場合、その並列処理で入れた記事データは、「記事のタイトルは存在するが本文が存在しない」状態になってしまうのです。

一応以下のようにすることで、各並列処理内ではTransactionを効かせることが出来ます。

並列処理で、以下の処理を行う
1. Transaction開始
2. DBに記事のタイトルデータを入れる
3. DBに記事の本文データを入れる
4. DBに記事の公開日データを入れる
5. Transaction終了

public void createArticleListInParallel() {
    final List<Article> articleList = List.of(
        new Article("タイトル", "本文", LocalDateTime.now()),
        // ...
    );

    articleList
            .parallelStream()
            .forEach(article -> this.createArticle(article));
}

@Transactional // 注：Transactionalは本来はprivateメソッドでは使えませんが、説明上付けています
private void createArticle(Article article) {
    this.setTitle(article.getTitle());
    this.setBody(article.getBody());
    this.setPublishDate(article.getPublishDate());
}