概要

Springの非同期処理を使ってリポジトリのクエリを非同期に実行することできると知ったので実装方法を調べました。
この記事は非同期クエリの簡単な実装例とその動作結果をまとめたものになります。

環境

Java 1.8.0_144
Spring Boot 1.5.9
- Spring Framework 4.3.13
MySQL 5.7.9

参考

[Spring Data JPA - Reference Documentation - 3.4.7. Async query results] (https://docs.spring.io/spring-data/jpa/docs/current/reference/html/#repositories.query-async)
[Class ThreadPoolTaskExecutor] (https://docs.spring.io/spring/docs/4.3.13.RELEASE/javadoc-api/org/springframework/scheduling/concurrent/ThreadPoolTaskExecutor.html)

データベース側の準備

動作確認を行いやすくするために検索に時間のかかるビューを作成します。

CREATE OR REPLACE VIEW async_test_view (
      id
    , sleep
    , create_at ) AS
SELECT MD5(UUID()) AS id
    , SLEEP(10) AS sleep
    , NOW() AS create_at
;

このビューを検索すると結果が返ってくるまで約10秒掛かります。

> select * from pseudo_delay_view;
+----------------------------------+-------+---------------------+
| id                               | sleep | create_at           |
+----------------------------------+-------+---------------------+
| da863db6ff1b064ebff03f00efdd224b |     0 | 2017-12-23 17:27:08 |
+----------------------------------+-------+---------------------+
1 row in set (10.00 sec)

Spring Boot側の実装

非同期処理の有効化とスレッドプールの設定

EnableAsyncアノテーションを付与して非同期処理を有効化します。
非同期処理の有効化にあたって必須ではありませんが、この例ではスレッドプールを使うように実装します。
タスクのキューサイズを指定します。キューに貯められたタスクをアイドルのスレッドが処理していきます。ThreadPoolTaskExecutorのデフォルトはInteger.MAX_VALUEです。
CorePoolSizeでプールサイズを指定します。MaxPoolSizeが上限数です。
CorePoolSize以上でアイドルのスレッドの生存時間を指定します。ThreadPoolTaskExecutorのデフォルトは60秒です。

@SpringBootApplication
// 1
@EnableAsync
public class DemoGradleApplication {
  public static void main(String[] args) {
    SpringApplication.run(DemoGradleApplication.class, args);
  }

  // 2
  @Bean
  public TaskExecutor taskExecutor() {
    ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();

    // 3
    executor.setQueueCapacity(2);
    // 4
    executor.setCorePoolSize(2);
    executor.setMaxPoolSize(3);
    // 5
    executor.setKeepAliveSeconds(10);

    executor.afterPropertiesSet();

    return executor;
  }

}

プールサイズの上限に達した場合

要求数がMaxPoolSize + QueueCapacity以上になった場合、デフォルトではRejectedExecutionExceptionという例外がスローされます。

要求が上限に達した場合のリジェクト処理に任意の処理を実行したい場合はRejectedExecutionHandlerインターフェースを実装したクラスを下記のようにThreadPoolTaskExecutorへ渡します。
明示しない場合のデフォルトはThreadPoolExecutor#AbortPolicyです。（RejectedExecutionException例外をスローします）

executor.setRejectedExecutionHandler((r,e) -> {
  // 実行したい任意の処理を実装
  throw new RejectedExecutionException("Task:" + r.toString() + " rejected from " + e.toString());
});

[インタフェースRejectedExecutionHandler] (https://docs.oracle.com/javase/jp/8/docs/api/java/util/concurrent/RejectedExecutionHandler.html)

エンティティ

ビューに対応するエンティティの実装です。特記事項はありません。

@Entity
@Table(name="pseudo_delay_view")
@Data
@AllArgsConstructor
@NoArgsConstructor
public class PseudoDelay implements Serializable {

    private static final long serialVersionUID = -9169553858944816379L;

    @Id
    private String id;
    @Column(name="sleep", nullable=false)
    private Integer sleep;
    @Column(name="create_at", nullable=false)
    private LocalDateTime createAt;

}

リポジトリ

クエリを発行するリポジトリの実装です。クエリを非同期で実行するための実装を行います。と言ってもメソッドにAsyncアノテーションを付与するだけです。

クエリを非同期で実行するメソッドにAsyncアノテーションを付与します。また戻り値の型にCompletableFutureを使います。
こちらは比較用の同期で実行するメソッドです。

public interface PseudoDelayRepository extends JpaRepository<PseudoDelay, String> {

    // 1
    @Async
    @Query("SELECT p FROM PseudoDelay AS p")
    CompletableFuture<PseudoDelay> findAsync();

    // 2
    @Query("SELECT p FROM PseudoDelay AS p")
    PseudoDelay findSync();

}

この例ではCompletableFutureを使っていますが、この他にFutureやSpringのListenableFutureが使えます。詳しくは[Spring Data JPA - Reference Documentation - 3.4.7. Async query results] (https://docs.spring.io/spring-data/jpa/docs/current/reference/html/#repositories.query-async)で確認できます。

[クラスCompletableFuture] (https://docs.oracle.com/javase/jp/8/docs/api/java/util/concurrent/CompletableFuture.html)

クエリが非同期に実行されているか確認する

確認用のサービスクラスの実装は下記の通りです。

非同期クエリの動作確認用メソッドです。
比較するための同期クエリの確認用メソッドです。

@Service
@Slf4j
public class AsyncTestServiceImpl implements AsyncTestService {

  private PseudoDelayRepository repository;

  public AsyncTestServiceImpl(PseudoDelayRepository repository) {
    this.repository = repository;
  }

  // 1
  @Transactional(readOnly = true)
  @Override
  public PseudoDelay async() {
    log.debug("start async");

    CompletableFuture<PseudoDelay> future = repository.findAsync();

    // 非同期でクエリを実行している間に何か処理を実行する
    log.debug("execute somethings");

    PseudoDelay result = null;
    try {
      // クエリの実行結果を受け取る
      result = future.thenApply(res -> {
        log.debug("async result : {}", res);
        return res;
      })
      .get();

      } catch (InterruptedException | ExecutionException e) {
        throw new RuntimeException(e);
      }
    }

    log.debug("end async");
    return result;
  }

  // 2
  @Transactional(readOnly = true)
  @Override
  public PseudoDelay sync() {
    log.debug("start sync");

    PseudoDelay result = repository.findSync();
    log.debug("sync result : {}", result);

    log.debug("end sync");
    return result;
  }

}

非同期クエリの動作確認

クエリを発行した直後に他の処理("*** execute somethings ***")が実行されていることが確認できます。

2017-12-23 19:55:36.194 DEBUG 5304 --- [nio-9000-exec-4] : start async
2017-12-23 19:55:36.195 DEBUG 5304 --- [nio-9000-exec-4] : *** execute somethings ***
2017-12-23 19:55:46.198 DEBUG 5304 --- [ taskExecutor-2] : async result : PseudoDelay(id=9904388341a9d8dbdfb230fb5b675224, sleep=0, createAt=2017-12-23T19:55:36)
2017-12-23 19:55:46.199 DEBUG 5304 --- [nio-9000-exec-4] : end async

同期クエリの動作確認

クエリの結果が返るまでブロックされていることが確認できます。

2017-12-23 19:57:49.465 DEBUG 5304 --- [nio-9000-exec-8] : start sync
2017-12-23 19:57:59.467 DEBUG 5304 --- [nio-9000-exec-8] : sync result : PseudoDelay(id=3a19a242c0207cd9ddad551ec2ccae66, sleep=0, createAt=2017-12-23T19:57:49)
2017-12-23 19:57:59.467 DEBUG 5304 --- [nio-9000-exec-8] : end sync

タイムアウトを指定する

タイムアウトの指定が可能です。指定した時間内に完了しなかった場合、TimeoutException例外がスローされます。

result = future.get(5, TimeUnit.SECONDS);

トランザクションタイムアウトについて

トランザクションにタイムアウトを設定し、そのタイムアウト以上に時間のかかるクエリを非同期に実行した場合、どのような結果になるか確認しました。
今回はトランザクションタイムアウトに5秒を設定し、上記の例にあるように10秒かかるクエリを非同期、同期の両方で実行してみました。

@Transactional(readOnly = true, timeout = 5)

非同期クエリの場合

タイムアウトを超えても例外は起きず、期待した結果にはなりませんでした。
原因はわかりませんがトランザクション内で別スレッドで実行するとトランザクション管理外になるようです。
オフィシャルのドキュメントに記載がないか調べたのですが調べ切れていません。
一応、それらしい記事に[Spring @Async and transaction management] (https://dzone.com/articles/spring-async-and-transaction)がありました。

同期クエリの場合

JDBCドライバの実装により変わってくると思いますがMySQLの場合は下記のエラーになります。

2017-12-23 20:17:18.297 ERROR 2260 --- [nio-9000-exec-1] o.h.engine.jdbc.spi.SqlExceptionHelper   : Statement cancelled due to timeout or client request
2017-12-23 20:17:18.319 ERROR 2260 --- [nio-9000-exec-1] o.a.c.c.C.[.[.[.[dispatcherServlet]      : Servlet.service() for servlet [dispatcherServlet] in context with path [/app] threw exception [Request processing failed; nested exception is org.springframework.orm.jpa.JpaSystemException: could not extract ResultSet; nested exception is org.hibernate.exception.GenericJDBCException: could not extract ResultSet] with root cause

com.mysql.jdbc.exceptions.MySQLTimeoutException: Statement cancelled due to timeout or client request
	at com.mysql.jdbc.MysqlIO.sqlQueryDirect(MysqlIO.java:2827) ~[mysql-connector-java-5.1.44.jar:5.1.44]

/... 省略