androidアプリ開発　kotlin部 : KotlinでRxJavaを利用して非同期、並列処理をする方法

　RxJavaは挙動を把握するのが難しいです。

RxJavaは1.xと2.xがあるけど、今は2.xが多くなってきたのでこっちを取り上げる。
　RxJavaは何やらいろいろな仕組みが詰まったライブラリだけど、リスト処理と非同期処理などを行うのに使うと便利そうだ。だが、リスト処理に関してはKotlinはJavaより優れているので、RxJavaをリスト処理目的で利用するのは最小限で済みそう。
　例えばRxJavaを導入することで「1から10までの値を渡し、偶数だけにフィルタリングしたうえ、値を10倍にして、ログ出力する」ことがメソッドチェーンを利用して1行でできるから素晴らしいというようなことが記述されていたが、kotlinではこんな風に記述できる。

var intList: List<Int> = arrayListOf(1, 2, 3, 4, 5, 6, 7, 8, 9, 10)
intList.filter { it % 2 == 0 }.forEach { Log.v("nullpo", (it * 2).toString()) }

　ので、ここでは非同期、並列処理について記述をしてみる。

■　RxJava2.xを使用するには

　gradleに下記の設定をする。必要に応じてバージョンは設定しておく。


dependencies {
    implementation 'io.reactivex.rxjava2:rxjava:2.1.14'
    implementation 'io.reactivex.rxjava2:rxandroid:2.0.2'
    implementation 'io.reactivex.rxjava2:rxkotlin:2.2.0'
}

■　簡単な非同期実行の方法1

　RxJavaは難しいので、まずは最も簡単なCompletableでのやり方から。
　Completableは失敗したか、完了したかのどちらかを返す機能を持っており、簡単な非同期処理を実現できる。例えば画像をダウンロードして保存するとか、WebAPIにPOSTするとか成否がわかればいいような処理に使われる。
　具体的には下記のソースコードのコメントを読んでもらえればわかりやすいと思う。
　まずは処理を記述して、成功だったらonCompleteを、エラーが発生したらonErrorを実行する。するとコールバックされて、それぞれ記述した処理が実行される。どのスレッドで動作させるかは指定できるので指定しておく。
　たったこれだけでメインスレッドとは別のスレッドで処理を実行させることができる。


class MainActivity : AppCompatActivity() {

    override fun onCreate(savedInstanceState: Bundle?) {
        super.onCreate(savedInstanceState)
        setContentView(R.layout.activity_main)

        //スレッド名を表示させる。もちろんここはメインスレッドになる
        Log.v("null", Thread.currentThread().name)

        //createの中のラムダ式を指定したスレッドで動作させることができる
        Completable.create {
            
            //ここが指定したスレッドで動作させられる
            //メインスレッド以外を指定すれば非同期、並列処理になる
            try {

                //別スレッドで動作していることがわかる
                Log.v("null", Thread.currentThread().name)

                //処理が完了したらonCompleteを呼んで、終了処理を行える
                it.onComplete()
            } catch (e: Exception) {

                //エラーが発生したらエラー情報を送ることができる
                it.onError(Throwable("error"))
            }
        }

                //別スレッドで動作させることを指定する
                .subscribeOn(Schedulers.newThread())
                .subscribe({

                    //it.onCompleteが呼ばれると実行される
                    Log.v("null", "2" + Thread.currentThread().name)
                }, {

                    //it.onErrorが呼ばれると実行される
                })
    }
}

■　簡単な非同期実行の方法2　コールバックをメソッドチェーンに記述したい場合

　subscribeにコールバックを記述できるが、メソッドチェーンに記述して可読性を良くすることもできる。
　ただし、RxJavaの問題かどうかはわからないが、it.onErrorを実行するときにはonError時の処理を引数としてもっているsubscribeを実行しないとエラーになる。なのでこのようにメソッドチェーンに記述する場合でも、subscribeには空の処理を入れておくこと。
　また一部のCompletableなどに関する設定を行うとき、チェーンメソッドに記述した設定では反映されないことがあるので、RxJavaの設定などに関する処理を記述する場合にはsubscribeのほうに記述をするほうが無難だろう。


class MainActivity : AppCompatActivity() {

    override fun onCreate(savedInstanceState: Bundle?) {
        super.onCreate(savedInstanceState)
        setContentView(R.layout.activity_main)

        Log.v("null", Thread.currentThread().name)
        Completable.create {
            try {
                Log.v("null", Thread.currentThread().name)
                it.onComplete()
            } catch (e: Exception) {
                it.onError(Throwable("error"))
            }
        }
                .subscribeOn(Schedulers.newThread())

                //onComplete時にコールバックされる
                .doOnComplete {
                    Log.v("null", "2" + Thread.currentThread().name)
                }

                //onError時にコールバックされる
                .doOnError {

                }

                //onErrorを実行したとき、onErrorの処理を記述する引数のメソッドを指定しないとエラーになる
                //とりあえず空のラムダ式を入れておけば問題ない
                .subscribe({ }, { })
    }
}

■　簡単な非同期実行の方法3　コールバックをメインスレッドで行いUI部品を操作する場合

　observeOnメソッドを利用すると、動作させるスレッドを指定することができる。このようにメソッドチェーンで記述すると、observerOn以降に記述されたメソッドチェーンのメソッドについては、observerOnで指定したスレッドで動作する。下記のようにすると、onCompleteはメインスレッドで動作するので、UI部品に対して処理が終了したことを伝えることができる。


class MainActivity : AppCompatActivity() {

    override fun onCreate(savedInstanceState: Bundle?) {
        super.onCreate(savedInstanceState)
        setContentView(R.layout.activity_main)

        Log.v("null", Thread.currentThread().name)
        Completable.create {

            //ここは別スレッドで動作する
            try {
                Log.v("null", Thread.currentThread().name)
                it.onComplete()
            } catch (e: Exception) {
                it.onError(Throwable("error"))
            }
}
                .subscribeOn(Schedulers.newThread())

                //これ以降のチェーンメソッドについてはメインスレッドで動作する
                .observeOn(AndroidSchedulers.mainThread())
                .doOnComplete {

                    //メインスレッドで動作する
                    //UI部品などの操作が可能
                    Log.v("null", "2" + Thread.currentThread().name)
                }
                .doOnError {
                    //メインスレッドで動作する
                }
                .subscribe({ }, { })
    }
}

　スレッドはその他にも、以下のようなものがあり、それぞれ指定できる。

AndroidSchedulers.from(looper) 特定のスレッドの指定
Schedulers.computation() 計算処理などの重い処理用用
Schedulers.io() データベースやファイルへの書き込み用
Schedulers.trampoline() 現在のスレッドに処理をプールしてマルチタスク的に一つずつ実行する

■　非同期で処理をしてデータをコールバックする方法

　これまでのCompletableでは、実行の結果、成功したか失敗したかという情報しか得られなかった。しかし、例えばWebAPIをたたいてJSON形式のデータを取得してコールバックさせたり、画像を取得してコールバックさせたいということもあると思う。そういうときには、下記に示すようにSingleを利用すると、データを引数に持たせてコールバックさせることができる。
　下記ではStringにしているが、bitmapなどにすれば画像を非同期でダウンロードし、UI部品であるimageViewなどに画像を表示させるようなことができる。


class MainActivity : AppCompatActivity() {
    override fun onCreate(savedInstanceState: Bundle?) {
        super.onCreate(savedInstanceState)
        setContentView(R.layout.activity_main)

        //メインスレッド
        Log.v("null", Thread.currentThread().name)

        //処理した結果を返すデータの型をジェネリクスで指定する
        //今回はhtmlを取得すると課程してStringを指定
        Single.create<String> {
            try {

                //ここは別スレッドで動作するので様々な処理をする
                Log.v("null", Thread.currentThread().name)

                //onSuccessで処理した結果を返す
                //例えばhttpでデータを取得してその取得したデータを返す
                it.onSuccess("結果")
            } catch (e: Exception) {
                it.onError(Throwable("失敗"))
            }
        }
                .subscribeOn(Schedulers.newThread())

                //これ以降はメインスレッドで動作する
                .observeOn(AndroidSchedulers.mainThread())
                .doOnSuccess {

                    //メインスレッド
                    Log.v("null", Thread.currentThread().name)

                    //取得したデータはラムダ式の省略でitに入っている
                    //"結果"がlogに出力される
                    Log.v("null",it)
                }
                .doOnError {

                }
                .subscribe({}, {})
    }
}

■　非同期で継続的なデータをコールバックする方法

　これまでは成功したか失敗したかか、成功した結果を通知するか失敗したかという通知の仕方だけだった。しかし処理によっては逐次様々なデータが送られてきたり、処理の途中経過を返したいような処理をしたい場合もあるだろう。例えば、大きなファイルをダウンロードしていて、その途中経過を知らせたいとか、計算結果の途中結果を知らせたいとか、GPSデータのように次々とデータが送られるので処理したいような場合に利用する。
　逐次なデータを扱うにはObsavableを利用する。すでに紹介した方法と同じように処理するが、違うところはonNextを何回も実行できるので、ここで逐次データを送って処理を行う。そして完了したらonCompleteを実行することで処理が終わったことを知らせることができる。
　具体的な流れはコメントとして記述しているので読んで欲しい。


class MainActivity : AppCompatActivity() {

    override fun onCreate(savedInstanceState: Bundle?) {
        super.onCreate(savedInstanceState)
        setContentView(R.layout.activity_main)

        Observable.create<String> {

            //ここは別スレッドで動作する
            Log.v("nullpo", Thread.currentThread().name)

            try {
                //何回もデータを送信できる
                //ここで逐次に発生するデータを送る
                it.onNext("ぬるぽ")
                it.onNext("ぬるぽ")
                it.onNext("ぬるぽ")

                //成功したらonComplete
                it.onComplete()
            } catch (e: Exception) {
                //エラーならonErrorを実行する
                it.onError(Throwable("しっぱい"))
            }
        }
                .subscribeOn(Schedulers.newThread())

                //これ以降はメインスレッドで実行する
                .observeOn(AndroidSchedulers.mainThread())

                //onNextが実行するたびに呼ばれる（他が1回に対して任意の回数呼べる）
                //ラムダ式のitにデータが入っているので、そのデータに対して処理を行う
                .doOnNext {
                    //メインスレッドで実行されるのでUI部品をいじれる
                    Log.v("nullpo", Thread.currentThread().name)
                    Log.v("null", "onNext")
                }

                //onErrorが実行されると呼ばれる
                .doOnError {
                    Log.v("nullpo", Thread.currentThread().name)
                    Log.v("null", "onError")
                }

                //onCompleteが実行されると呼ばれる
                .doOnComplete {
                    Log.v("nullpo", Thread.currentThread().name)
                    Log.v("null", "onComplete")
                }

                //onErrorを実行するときにはsubscribeでError処理をする引数のあるsbscripeを呼ばないと落ちる
                //なのでとりあえず何もしないラムダ式を記述しておく
                .subscribe({}, {})
    }
}

名前
	情報を記憶評価顔星

androidアプリ開発 kotlin部

KotlinでRxJavaを利用して非同期、並列処理をする方法

コメントする

androidアプリ開発　kotlin部