SpringBootの@AsyncとCompletableFutureを組み合わせて並列ロジックを作る

# はじめに
SpringBootでは非同期処理を簡単に使える仕組みが用意されています。 非同期処理をさせたいメソッドに `@Async` をつけ、アプリケーション全体の設定として `@EnableAsyc` をつけるだけで良いのですが、これはControllerだけではなく、他のステレオタイプでも実施できます。

これにより、1つのController＋Serviceから複数のデータソース・複数環境のデータを結合して返せるようになります。

簡単な例として、データベースの検索結果を取得するRepository、もう一方はWebAPIからJSONで検索結果をもらうRepositoryから得られた結果を結合して返す機能を作るとします。
この2つを結合した結果を返す、一番単純で簡単な方法、2つのRepositoryの結果を順次呼び出して、その結果を返すことです。

データベースの検索を行うRepositoryをA、WebAPIを実行するRepositoryをBとして、

Aが結果を返すまでに1秒かかるとし、
Bが結果を返すまでに2秒かかるとします。

Aが終わったあとにBを順次呼び出す方法では、検索結果が得られるまでに A+Bの3秒かかります。

かたや、AとBが並列に動作すれば、検索結果が得られるまで、最大2秒で済むでしょう。これを簡単に実現できるのが、`@Async`とCompletableFutureを組み合わせる方法です。

## 実装例

### Service
ServiceクラスからRepositoryを呼ぶときに、並列で実行したい処理に対してCompletableFutureで受け取り、その後、結合したいタイミングで、CompletableFuture.allOf(並列処理した結果....).join() をします。

```java:MainService.java
import java.util.List;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import java.util.stream.Collectors;

import org.springframework.stereotype.Service;

import com.github.apz.springsample.dto.Item;
import com.github.apz.springsample.dto.Order;
import com.github.apz.springsample.dto.OrderedItem;
import com.github.apz.springsample.repository.ItemRepository;
import com.github.apz.springsample.repository.OrderRepository;

import lombok.RequiredArgsConstructor;
import lombok.extern.log4j.Log4j2;

@Service
@RequiredArgsConstructor
@Log4j2
public class MainService {
	private final ItemRepository itemRepository;
	private final OrderRepository orderRepository;

	public List<OrderedItem> selectAllItems() throws InterruptedException, ExecutionException {
		CompletableFuture<List<Item>> items = itemRepository.allItems();
		CompletableFuture<List<Order>> orders = orderRepository.getRecentOrders();

		CompletableFuture.allOf(items, orders).join();
		log.info("MainService : merged.");

		List<Item> itemList = items.get();
		List<Order> orderList = orders.get();

		// 結合処理をして返す
		return getOrderedItem(itemList, orderList);
	}
}
```

### Repository
それぞれのRepositoryでは、@Asyncを付与して、CompletableFutureをラップした結果を返すようにします。

ItemRepositoryは1秒待機したあとデータベースに検索します（Thread.sleepを使って、仮想的に待機時間を作っています）

```java:ItemRepository.java
import java.util.List;
import java.util.concurrent.CompletableFuture;

import org.mybatis.spring.SqlSessionTemplate;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Repository;

import com.github.apz.springsample.dto.Item;

import lombok.RequiredArgsConstructor;
import lombok.extern.log4j.Log4j2;

@Repository
@RequiredArgsConstructor
@Log4j2
public class ItemRepository {
	private final SqlSessionTemplate sqlSessionTemplate;

	@Async
	public CompletableFuture<List<Item>> allItems() throws InterruptedException {
		log.info("ItemRepository >>");
		Thread.sleep(1000);  // 1秒待機

		// データベースから検索結果を取得
		List<Item> result = sqlSessionTemplate.selectList("selectItems");
		log.info("<< ItemRepository");
		return CompletableFuture.completedFuture(result);

	}
}
```

```Java:OrderRepository.java
import java.time.LocalDateTime;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CompletableFuture;

import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Repository;

import com.github.apz.springsample.dto.Order;

import lombok.RequiredArgsConstructor;
import lombok.extern.log4j.Log4j2;

@Repository
@RequiredArgsConstructor
@Log4j2
public class OrderRepository {
	@Async
	public CompletableFuture<List<Order>> getRecentOrders() throws InterruptedException {
		log.info("OrderRepository >>");
		Thread.sleep(2000);  // 2秒待機
		List<Order> orders = getRecentOrderData();
		log.info("<< OrderRepository");
		return CompletableFuture.completedFuture(orders);
	}
}
```

実際にこのServiceを実行してみると、

```
2018-07-22 22:03:33.613  INFO 11432 --- [    AsyncTask-2] c.g.a.s.repository.OrderRepository       : OrderRepository >>
2018-07-22 22:03:33.613  INFO 11432 --- [    AsyncTask-1] c.g.a.s.repository.ItemRepository        : ItemRepository >>
2018-07-22 22:03:34.615  INFO 11432 --- [    AsyncTask-1] c.g.a.s.repository.ItemRepository        : << ItemRepository
2018-07-22 22:03:35.613  INFO 11432 --- [    AsyncTask-2] c.g.a.s.repository.OrderRepository       : << OrderRepository
2018-07-22 22:03:35.613  INFO 11432 --- [nio-8080-exec-1] c.g.a.springsample.service.MainService   : MainService : merged.
```

OrderRepositoryとItemRepositoryがほぼ同時に実行され、ItemRepositoryが1秒後、OrderedRepositoryが2秒後に結果を返し、MainServiceはすべてのReositoryが返事を返した時間、つまり2秒後にその後の処理を実行されました。

## その他必要な設定

@Asyncを使うときは、SpringBootの設定にて、@EnableAsyncを付与します。

```java:SpringAnyncServiceApplication.java
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.scheduling.annotation.EnableAsync;

@SpringBootApplication
@EnableAsync
public class SpringAnyncServiceApplication {

	public static void main(String[] args) {
		SpringApplication.run(SpringAnyncServiceApplication.class, args);
	}
}
```

また、CompletableFutureの同時スレッド数や上限を以下のように設定可能です。

```java:AsyncConfiguration.java
import java.util.concurrent.Executor;

import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;

@Configuration
public class AsyncConfiguration {
    @Bean
    public Executor asyncExecutor() {
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        executor.setCorePoolSize(2);
        executor.setQueueCapacity(2);
        executor.setMaxPoolSize(40);
        executor.setThreadNamePrefix("AsyncTask-");
        executor.initialize();
        return executor;
    }
}
```

## 余談
データベースを使った検索をする際に、1つの巨大なSQLで実現せずに複数のSQLに分割して実行する場合にもこの手法は非常に有効ですが、データソースの同時接続数を多くしないとあまり意味がなく、データベースの負荷は当然高まります。

## 参考資料・参考文献

[https://spring.io/guides/gs/async-method/](https://spring.io/guides/gs/async-method/)
[blog.ik.am : 誤解しがちなThreadPoolTaskExecutorの設定](https://blog.ik.am/entries/443)