Trong bài viết này, chúng ta hãy cùng tìm hiểu cơ bản về RxJava Disposables.
Reactive Android Programming bao gồm RxJava, RxAndroid và RxKotlin.
RxJava là thư viện Reactive Programming được sử dụng nhiều nhất trong thế giới Phát triển Android. Nó hỗ trợ quản lý đa luồng tốt hơn và làm cho mã trực quan hơn.
RxAndroid là một mô-đun nhẹ, liên kết các thành phần cụ thể của Android với các lớp RxJava. RxKotlin giúp viết các phương thức RxJava dễ dàng hơn bằng cách cung cấp các extension functions thuận tiện.
Disposables
Disposables là gì? Disposables trong tiếng Anh ngụ ý sự thuận tiện ngắn hạn. Điều này cũng có nghĩa là chúng tồn tại trong thời gian ngắn hoặc có nghĩa là bị loại bỏ sau khi sử dụng. Ý tưởng tương tự cũng được chuyển tải trong Disposables của RxJava.
Khi một Observer đăng ký vào một Emitter hoặc Observables, bạn tạo một stream. Stream này chiếm tài nguyên (resources) mà sau này trở thành disposable “solid waste”. Bạn cần xử lý nó nếu không luồng sẽ chạy trong một thời gian dài.
Observable có một phương thức gọi là onComplete() sẽ thực hiện việc xử lý cho bạn khi được gọi. Tuy nhiên, nhiều lúc, bạn sẽ thấy có lợi và thuận tiện hơn khi có khả năng hủy đăng ký của mình một cách dễ dàng và bất cứ lúc nào.
Bây giờ, bạn sẽ giải quyết một trường hợp phức tạp hơn. Stream sẽ chạy vô tận và bạn sẽ sử dụng Disposables để xử lý chúng để tránh rò rỉ bộ nhớ (memory leaks).
Disposable là một stream hoặc một liên kết giữa một Observable và Observer. Kiểm tra nhanh tài liệu cho thấy rằng nó có hai phương thức chính là dispose() và isDisposed(). Cái trước xử lý liên kết, trong khi cái sau kiểm tra nếu liên kết đã được xử lý.
Testing Disposables
Khi bạn thiết lập đăng ký giữa Observable và Observer, nó sẽ trả về một Disposable. Ví dụ, hãy xem đoạn mã sau:
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 | import android.util.Log import io.reactivex.Observable import java.util.concurrent.TimeUnit object DisposableTester { @JvmStatic fun main() { val seconds = Observable.interval(1, TimeUnit.SECONDS) val disposable = seconds.subscribe({ l -> logData(l) }) //sleep 10 seconds sleep(10000) //Dispose and stop emissions disposable.dispose() Log.d("Test","Disposed!") //Sleep 10 seconds to prove //There are no more emissions sleep(10000) } private fun logData(l: Long) { Log.d("Test","Received: " + l) } private fun sleep(millis:Int) { try { Thread.sleep(millis.toLong()) } catch (e:InterruptedException) { e.printStackTrace() } } } |
Một Observable chạy và phát ra mỗi giây. Sau mười giây phát ra, tài nguyên Disposable trả về từ subscribe() được xử lý bằng cách gọi rõ ràng dispose(). Sau đó, bộ đếm thời gian mười giây khác phát ra để xác minh rằng tài nguyên đã được xử lý.
Xử lý Disposables bên ngoài là một cách để xử lý các tài nguyên không còn cần thiết. Kể từ RxJava 2.0, Observer có khả năng loại bỏ đăng ký bất cứ lúc nào. Ví dụ:
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 | import io.reactivex.Observer import io.reactivex.disposables.Disposable object DisposableTester { var myObserver: Observer<Int> = object: Observer<Int> { private var disposable: Disposable? = null override fun onSubscribe(disposable: Disposable) { this.disposable = disposable } override fun onNext(value:Int) { //Has access to Disposable } override fun onError(e:Throwable) { //Has access to Disposable } override fun onComplete() { //Has access to Disposable } } } |
Nếu bất cứ lúc nào phát thải không còn cần thiết trong onNext(), onError() hoặc onComplete(), bạn có thể dừng chúng.
CompositeDisposables
Khi phát triển ứng dụng, có những kịch bản mà bạn cần nhiều hơn một subscription. Lấy dữ liệu trực tiếp từ nhiều nguồn trong một ứng dụng du lịch cho khách sạn, các tour du lịch và vé máy bay là một ví dụ tuyệt vời.
Bạn cần sử dụng CompositeDisposables để quản lý tài nguyên. Nó implement Disposable và sau đó giữ một bộ sưu tập các disposables. Xem ví dụ sau:
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 | import android.util.Log import io.reactivex.Observable import io.reactivex.disposables.CompositeDisposable import java.util.concurrent.TimeUnit object DisposableTester { private val disposables = CompositeDisposable() @JvmStatic fun main() { val seconds = Observable.interval(1, TimeUnit.SECONDS) //Subscribe and capture disposables val disposable1 = seconds.subscribe({ l -> logData(l, 1) }) val disposable2 = seconds.subscribe({ l -> logData(l, 2) }) //Put both disposables into CompositeDisposable disposables.addAll(disposable1, disposable2) //Sleep 10 seconds sleep(10000) //Dispose all disposables disposables.dispose() Log.d("Test", ("All Disposed!")) //Sleep 10 seconds to prove //there are no more emissions sleep(10000) } private fun logData(l: Long, observerNumber: Int) { Log.d("Test", ("Observer " + observerNumber + ": " + l)) } private fun sleep(millis:Int) { try { Thread.sleep(millis.toLong()) } catch (e:InterruptedException) { e.printStackTrace() } } } |
Trong triển khai này, mã được sử dụng trên tiện ích đơn giản này giúp bạn quản lý một bộ sưu tập các Disposables. Bằng cách gọi add() hoặc addAll(), bạn có thể loại bỏ tất cả chúng cùng một lúc khi chúng không còn cần thiết nữa.
Xin chúc mừng! Qua đây bạn đã làm mới kiến thức của mình về những điều cơ bản của Reactive Programming và học về Disposables.
ref: https://www.raywenderlich.com/3983802-working-with-rxjava-disposables-in-kotlin#toc-anchor-001