CoroutinesのlaunchとasyncとChannelとFlowの使い分け

# はじめに
Kotlinでは非同期処理（スレッド）の実装方法が色々でてきました。
Androidアプリ開発をやっている私としてはKotlinでアプリ開発をすると、正直何を使って良いのかわからない。

<img src="https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/99960/84616ab5-01eb-035b-78a4-7eb35d3735b2.png" width="60%" />


今一度、詳細を調べてみて、使い分けをはっきりさせようと思います。


# 動作環境
この記事の動作環境は以下のとおりです。

Kotln：1.4.10


# 目標
以下を目標とします。

 ** Coroutinesによる非同期処理の使い分けをはっきりさせる** 

# 各種メソッドの定義
とりあえず、よく見かけるlaunch、async、flow関数（Flowビルダー）、Flow#collect、FlowCollector#emit、Channel#send、Channel#receiveの定義を確認してます。
[サイト](https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/index.html)に少しまとまっているのでそこからまとめてみます。

|メソッド|メソッドの定義|戻り値|概要|
|:---|:---|:---|:---|
|launch|fun CoroutineScope.launch(<br>　context: CoroutineContext = EmptyCoroutineContext,<br>　start: CoroutineStart = CoroutineStart.DEFAULT,<br>　block: suspend CoroutineScope.() -> Unit<br>): Job (source)|Job|任意の結果を返せないコルーチン|
|async|fun <T> CoroutineScope.async(<br>　context: CoroutineContext = EmptyCoroutineContext,<br>　start: CoroutineStart = CoroutineStart.DEFAULT,<br>　block: suspend CoroutineScope.() -> T<br>): Deferred<T> (source)|Deferred|任意の単一の結果を返す|
|Channel#send|abstract suspend fun send(element: E): Unit (source)|なし|Channelに結果を送信する。|
|Channel#receive|abstract suspend fun receive(): E (source)|Channelのインスタンス化した時にジェネリクスで指定した型|Channnelからデータを取得する。|
|flow関数（Flowビルダー）|fun <T> flow(<br>    block: suspend FlowCollector<T>.() -> Unit<br>): Flow<T> (source)|Flowオブジェクト|ビルダーでインスタンス化を生成する際にジェネリクスで指定した型のデータを返す|
|FlowCollector#emit|abstract suspend fun emit(value: T): Unit (source)|なし|Flowに値を送信する。|
|Flow#collect|abstract suspend fun collect(<br>　collector: FlowCollector<T><br>): Unit (source)|なし|Flowがデータを受け取るやつ。collect拡張関数で収集した値を出力する。|

とりあえず、並べてみたけど、一番最初の感想は、「よくわからない！！」です。
やっぱり、実際に動かして見ながら理解していきます。

# サンプルプロジェクト

[github](https://github.com/ktaro-nakaya/qiita-coroutines-channel-and-flow)からダウンロード可能です。

# launch
## 概要
launchをまず図にしてみました。
<img src="https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/99960/c3846163-ed45-1bb5-344b-fb1fe264de2f.png" width="80%" />




launchはスレッドを生成して、実行してくれます。
とうぜんメインの処理とは別スレッドになるので、launchしたスレッドが終わるまで待つ歯科差なさそうです。
スレッドの状態を確認するプロパティがあるので、うまく使って待ってあげる必要がありそうです。

状態を確認するプロパティは下記のようなものがあります。

|メソッド|説明|
|:--|:--|
|Job#isActive|キャンセルされておらず、スレッドが実行中の時にtrueを返す。|
|Job#isCancelled|スレッドがキャンセルされている時にtrueを返す。|
|Job#isCompleted|スレッドの実行が完了している時にtrueを返す。|


## サンプルコード
実際に、launchを使ったコードを確認します。
新しいスレッドでは、配列の文字列を標準出力で表示しています。

```kotlin:Launch.kt
package launch.basic

import kotlinx.coroutines.CoroutineScope
import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.delay
import kotlinx.coroutines.launch

val stories = arrayOf(
    "赤ずきんちゃんの４行ストーリ",
    "１．赤ずきんちゃん、おばあさんの家に行く",
    "２．赤ずきんちゃん、オオカミと遭遇、おばあさんオオカミに食べられる",
    "３．赤ずきんちゃん、オオカミに食べられる",
    "４．赤ずきんちゃん、猟師に助けられる"
)

fun main() {
    val job = CoroutineScope(Dispatchers.Default).launch {
        repeat(5) { count ->

            delay(500.toLong())

            println(stories[count])
        }
    }

    // Coroutineが終わるまで待つ
    while (!job.isCompleted) {
        Thread.sleep(100)
    }
}
```


# async
次は、よく見かけるasyncを確認していきます。
## 概要
asyncの特徴をまず図にしてみました。
<img src="https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/99960/7b2f3ab5-716f-6bd4-8a4a-41e4af8a817a.png" width="80%" />


launchとの違いは、戻り値を返せるということです。
戻り値の型に特に制限が無いため、任意の値を返せます。
また、launchの場合、処理が終わったかどうかをプロパティで判断していました。asyncの場合は、戻り値であるDeferredに処理が終わるまで待つメソッドが用意されています。また、キャンセルも行えます。協力的なキャンセルになっています。協力的なキャンセルについてはこの[記事](https://qiita.com/naoi/items/2149dd2cf6bf46f52521)を参照してください。

メソッドの終了と戻り値を受け取るメソッドは以下のとおりです。

|メソッド|説明|
|:--|:--|
|Deferred#await|asyncの処理が終わるまでまちます。|
|Deferred#getComplete|戻り値を取得できます。戻り値の型に制約はないので、任意の型を指定できます。|


## サンプルコード
### 基本

```Kotlin:AsyncBasic.kt
package async.basic

import kotlinx.coroutines.*
import launch.basic.stories

val stories = arrayOf(
    "赤ずきんちゃんの４行ストーリ",
    "１．赤ずきんちゃん、おばあさんの家に行く",
    "２．赤ずきんちゃん、オオカミと遭遇、おばあさんオオカミに食べられる",
    "３．赤ずきんちゃん、オオカミに食べられる",
    "４．赤ずきんちゃん、猟師に助けられる"
)

fun main() {
    val job = CoroutineScope(Dispatchers.Default).launch {
        repeat(5) { count ->

            val deffer = async {
                delay(500)
                stories[count]
            }

            println("${count + 1}回目のasyncが終わるまで待つ")
            deffer.await()
            println(deffer.getCompleted())
        }
    }

    // Coroutineが終わるまで待つ
    while (!job.isCompleted) {
        Thread.sleep(100)
    }
}
```
### cancel

続いて、キャンセルを実装したコードも確認しておきましょう。

```kotlin:AsyncCancel.kt
package async.cancel

import kotlinx.coroutines.*

val stories = arrayOf(
    "赤ずきんちゃんの４行ストーリ",
    "１．赤ずきんちゃん、おばあさんの家に行く",
    "２．赤ずきんちゃん、オオカミと遭遇、おばあさんオオカミに食べられる",
    "３．赤ずきんちゃん、オオカミに食べられる",
    "４．赤ずきんちゃん、猟師に助けられる"
)

fun main() {
    val job = CoroutineScope(Dispatchers.Default).launch {
        repeat(5) { count ->

            val deffer = async {
                // 2回目でキャンセル
                if (count == 2) {
                    cancel("キャンセル")
                }
                delay(500)
                launch.basic.stories[count]
            }

            println("${count + 1}回目のasyncが終わるまで待つ")
            deffer.await()
            println(deffer.getCompleted())
        }
    }

    // Coroutineが終わるまで待つ
    while (!job.isCompleted || !job.isCancelled) {
        Thread.sleep(100)
    }
}
```

# Channelの基本
いよいよ1.4で追加されたFlowやChannelについて説明していきます。
まずはChannelから確認していきます。

## 概要
channelをまず図にしてみました。
<img src="https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/99960/36c3b365-3de6-e074-21ee-b2ce456061b4.png" width="80%" />


Channelはホットストリームなんて言われています。
なんとなくのChannelのイメージですが、入れ物的なものでしょうか。
ChannelはメインからもCoroutineからでも使える入れ物で、単なる入れ物ではなく、取り出す（受信する）スレッドに待ってもらったり、Channelにデータを入れる（送信する）スレッドには、データが入っているので送信を待ってもらうなどの交通整理を行います。

Channelがホットストリームといわれる所以であり、特徴的として「 **受信者（値を取り出す処理）があるかどうかに関わらず送信処理（データを入れる）を行う。** 」ということです。

そのため以下の注意が必要です。

 **送信した回数分、受信しないとメモリーリークを起こす。言い換えると、受け取ってもらえるまで送信側の処理が動き続ける！！** 

送受信のメソッドについてサンプルコードを見る前に確認しましょう。

|メソッド|説明|
|:--|:--|
|Channel#send|Channelにデータを送信する。|
|Channel#recieve|Channelからデータを受け取る。|


## サンプルコード

```kotlin:ChannelBasic.kt
package channel.basic

import kotlinx.coroutines.channels.Channel
import kotlinx.coroutines.delay
import kotlinx.coroutines.launch
import kotlinx.coroutines.runBlocking

val stories = arrayOf(
    "赤ずきんちゃんの４行ストーリ",
    "１．赤ずきんちゃん、おばあさんの家に行く",
    "２．赤ずきんちゃん、オオカミと遭遇、おばあさんオオカミに食べられる",
    "３．赤ずきんちゃん、オオカミに食べられる",
    "４．赤ずきんちゃん、猟師に助けられる"
)

fun main() = runBlocking {
    // Channel生成（Stringのみを送受信できる）
    val channel = Channel<String>()

    // 非同期処理の起動
    launch {
        stories.forEach {

            // Channelに文字列を送信
            channel.send(it)
            println("送信：$it")
            // 1行待ち合わせ
            delay(1000)
        }
    }

    // Channelに５回送信されるため５回繰り返す
    repeat(5) {
        // Channelから受信
        val story = channel.receive()
        println("受信：$story")
    }

    // 他に下記のようにも記述可能
//    for (story in channel) {
//        // Channelから受信
//        val story = channel.receive()
//        println("受信：$story")
//    }

    println("おわり")
}
```

データの受取る方法として、forを使っても問題ありません。

```kotlin
    for (story in channel) {
        // Channelから受信
        val story = channel.receive()
        println("受信：$story")
    }
```

# ChannelのBuffer

## 概要
Channelはデータの入れ物と説明しましたが、これまでは単一のデータしか格納できませんでした。しかし、バッファーを設定することによって、複数のでーたを格納できます。
まず図にしてみました。
<img src="https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/99960/ea03c8c0-5978-6692-4099-299249af7637.png" width="80%" />

## サンプルコード
Channelをインスタンス化する際、コンストラクターに整数を渡すことによりバッファーを設定できます。


```kotlin:ChannelBuffer.kt
package channel.buffer

import kotlinx.coroutines.channels.Channel
import kotlinx.coroutines.delay
import kotlinx.coroutines.launch
import kotlinx.coroutines.runBlocking

val stories = arrayOf(
    "赤ずきんちゃんの４行ストーリ",
    "１．赤ずきんちゃん、おばあさんの家に行く",
    "２．赤ずきんちゃん、オオカミと遭遇、おばあさんオオカミに食べられる",
    "３．赤ずきんちゃん、オオカミに食べられる",
    "４．赤ずきんちゃん、猟師に助けられる"
)

fun main() = runBlocking {
    // Channel生成（Stringのみを送受信できる）
    val channel = Channel<String>(5)

    // 非同期処理の起動
    launch {
        stories.forEach {
            // Channelに文字列を送信
            channel.send(it)
            println("送信：$it")
        }
    }

    // Channelに５回送信されるため５回繰り返す
    repeat(5) {
        Thread.sleep(1000)
        // Channelから受信
        val story = channel.receive()
        println("受信：$story")
    }
    println("おわり")
}
```

# Channelのcloseとキャンセル

## 概要
Channelはメモリーリークしてしまう可能性があると説明しましたが、回避する方法として、closeやcacelがあります。
まず図にしてみました。
<img src="https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/99960/4ab31057-a4d4-df63-4858-42837cba3661.png" width="80%" />



それぞれ特徴が異なります。
メソッドの説明は以下のとおりです。

|メソッド|説明|
|:--|:--|
|Channel#canel|要素の受信をキャンセルします。Channelを閉じ、バッファリングされた送信済み要素をすべて削除します。キャンセルをすると例外(java.util.concurrent.CancellationException)が発生します。|
|Channel#close|Channelを閉じる。今後は呼び出してもfalseを返す。|


## サンプルコード

### cancel

```kotlin:ChannelCancel.kt
package channel.cancel


import kotlinx.coroutines.channels.Channel
import kotlinx.coroutines.delay
import kotlinx.coroutines.launch
import kotlinx.coroutines.runBlocking

val stories = arrayOf(
    "赤ずきんちゃんの４行ストーリ",
    "１．赤ずきんちゃん、おばあさんの家に行く",
    "２．赤ずきんちゃん、オオカミと遭遇、おばあさんオオカミに食べられる",
    "３．赤ずきんちゃん、オオカミに食べられる",
    "４．赤ずきんちゃん、猟師に助けられる"
)

fun main() = runBlocking {
    // Channel生成（Stringのみを送受信できる）
    val channel = Channel<String>()

    // 非同期処理の起動
    launch {
        stories.forEach {

            // Channelに文字列を送信
            channel.send(it)
            println("送信：$it")
            // 1行待ち合わせ
            delay(500)

            // 途中で中断
            if (stories.indexOf(it) == 2) {
                channel.cancel()
            }
        }
    }

    // Channelに５回送信されるため５回繰り返す
    repeat(5) {
        // Channelから受信
        val story = channel.receive()
        println("受信：$story")
    }
    println("おわり")
}
```

### close

```kotlin:ChannelClose.kt
package channel.close

import kotlinx.coroutines.channels.Channel
import kotlinx.coroutines.delay
import kotlinx.coroutines.launch
import kotlinx.coroutines.runBlocking

val stories = arrayOf(
    "赤ずきんちゃんの４行ストーリ",
    "１．赤ずきんちゃん、おばあさんの家に行く",
    "２．赤ずきんちゃん、オオカミと遭遇、おばあさんオオカミに食べられる",
    "３．赤ずきんちゃん、オオカミに食べられる",
    "４．赤ずきんちゃん、猟師に助けられる"
)

fun main() = runBlocking {
    // Channel生成（Stringのみを送受信できる）
    val channel = Channel<String>()

    // 非同期処理の起動
    launch {
        stories.forEach {

            // Channelに文字列を送信
            channel.send(it)
            println("送信：$it")
            // 1行待ち合わせ
            delay(500)

            // 途中で中断
            if (stories.indexOf(it) == 2) {
                channel.close()
            }
        }
    }

    // Channelに５回送信されるため５回繰り返す
    repeat(5) {
        // Channelから受信
        val story = channel.receive()
        println("受信：$story")
    }
    println("おわり")
}
```




# Flowの基本

## 概要
Flowはコールドストリームと呼ばれています。ホットストリームであるChannelと大きく動きが異なります。
まず図にしてみました。
<img src="https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/99960/20f16fb5-6f61-1e6c-4436-9c7d94d70e2b.png" width="80%" />

FlowはChannelと違い、受信する処理（collectメソッド）が確定していないと、送信処理（emitメソッド）を行いません。結果として、Flowのは実行されません。
そのため、メモリーリークは発生しません。
また、キャンセルは協力的なキャンセルを実装しています。
ここでは、基本的な呼び出しとキャンセルのサンプルコードを確認します。

## サンプルコード

### 基本的なFlowの使い方

```kotlin:FlowBasic.kt
package flow.basic

import kotlinx.coroutines.delay
import kotlinx.coroutines.flow.collect
import kotlinx.coroutines.flow.flow
import kotlinx.coroutines.launch
import kotlinx.coroutines.runBlocking

val stories = arrayOf(
    "赤ずきんちゃんの４行ストーリ",
    "１．赤ずきんちゃん、おばあさんの家に行く",
    "２．赤ずきんちゃん、オオカミと遭遇、おばあさんオオカミに食べられる",
    "３．赤ずきんちゃん、オオカミに食べられる",
    "４．赤ずきんちゃん、猟師に助けられる"
)

fun teller() = flow<String> {

    repeat(stories.count()) {
        Thread.sleep(1000)
        emit(stories[it])
        println("${it+1}回目のemitを呼び出したよ")
    }
}


fun main() = runBlocking<Unit> {
    launch {
        for (i in 1..3) {
            println("${i}mainメソッドで回目の遅延処理")
            delay(100)
        }
    }

    val collector = teller()

    collector.collect { value ->
        println(value)
        Thread.sleep(100)
    }
    println("おわり")
}
```

### 協力的なキャンセルを実装した例

```kotlin:FlowCancel.kt
package flow.cancel

import kotlinx.coroutines.delay
import kotlinx.coroutines.flow.collect
import kotlinx.coroutines.flow.flow
import kotlinx.coroutines.launch
import kotlinx.coroutines.runBlocking
import kotlinx.coroutines.withTimeoutOrNull

val stories = arrayOf(
    "赤ずきんちゃんの４行ストーリ",
    "１．赤ずきんちゃん、おばあさんの家に行く",
    "２．赤ずきんちゃん、オオカミと遭遇、おばあさんオオカミに食べられる",
    "３．赤ずきんちゃん、オオカミに食べられる",
    "４．赤ずきんちゃん、猟師に助けられる"
)

fun teller() = flow<String> {

    repeat(stories.count()) {
        Thread.sleep(1000)
        emit(stories[it])
        println("${it + 1}回目のemitを呼び出したよ")
    }
}


fun main() = runBlocking<Unit> {

    val collector = teller()

    // 2.5秒以上かかる場合は、キャンセルする
    withTimeoutOrNull(2500) {
        collector.collect { value ->
            println(value)
            Thread.sleep(100)
        }
    }
    println("おわり")
}
```


# まとめ
ここまで説明した内容をまとめた上で、使い分けを考えるとしたら、以下のようになります。

|種類|使い分け|
|:--|:--|
|launch|戻り値は不要でコルーチンを起動したいとき|
|async|任意の戻り値が必要で、複数の処理を並列で実行して、処理を待ち合わせたいとき|
|Channel|同じ型の複数の戻り値でリアルタイムで処理を行いとき|
|Flow|同じ型の複数の戻り値で必ず処理を完結させる必要があるとき|

となります。
Androidで考えるとlaunchでコルーチンを起動して、async、Channel、Flowを子コルーチンとして実行することが現実的な気がしてきました。