Giới thiệu

Xử lý luồng luôn là một chủ đề khó trong lập trình. Và nó cực kì quan trọng trong Android, nhất là khi ta phải đưa tất cả luồng xử lý network, database, tính toán… xuống tác vụ nền để tránh block UI.

Vì vậy, không có cách nào để để thoát được việc phải học nó cả. Nhưng yên tâm, mọi thứ sẽ trở nên đơn giản hơn với RxJava.

Rx Diagram

Trong bài viết trước, mình đã đề cập đến RxJava có thể được chia thành 3 phần đơn giản : Producer, Operators và Consumer.

Hãy tưởng tượng rằng, một số tác vụ tiêu tốn thời gian cần thực hiện tưới nền và chỉ Consumer chạy trên UI thread.

observeOn(…)

Để xác định xem chuỗi operators được thực hiện trên luồng nào, ta chỉ cần đơn giản gắn thêm chuỗi observeOn ứng với thread mong muốn.

Nó sẽ tương đương như sau:

1
2
3
4
5
6
7

producer().observedOn(Schedulers.computational())
          .operation1()
          .operation2()
          .observedOn(Schedulers.io())
          .moreOperations()
          .observedOn(AndroidScheduler.mainThread())
          .subscribe(...)

Nhìn khá simple đúng không ? :D

Có thể bạn sẽ đặt ra 1 câu hỏi…

Thế còn xử lý phía Producer thì sao ?

Giả sử hoạt động của producer rất chậm, và ta cũng muốn đưa nó vào background thì phải làm thế nào ?

1

Observable.fromCallable { superSlowProcess() }.subscribe{ ... }

subscribeOn(…)

Tương tự observeOn(…), sử dụng subscribeOn(…) nhưng nó khác ở điểm, bạn có thể subscribeOn(…) ở bất cứ đâu trong chuỗi ( sau Producer và trước Consumer).

Mã code sẽ trông như thế này:

1
2
3
4
5

Observable.fromCallable { superSlowProcess() }
          .operators // or operators...
          .subscribeOn(Schedulers.io())          
          .observeOn(AndroidScheduler.mainThread())
          .subscribe{ ... }

Hoặc bạn cũng có thể viết như bên dưới đây, cùng trả về một kết quả, chỉ là ta có thể dễ dàng thấy nới đăng kí và quan sát gần nhau hơn. Đây cũng là cách mình thường xuyên dùng ::lol::

1
2
3
4
5

Observable.fromCallable { superSlowProcess() }
          .operators // or operators...
          .subscribeOn(Schedulers.io())          
          .observeOn(AndroidScheduler.mainThread())
          .subscribe{ ... }

Conclution

OK. Hi vọng qua bài viết này các bạn có thể hiểu qua về cách xử lý luồng trong Rx. Bài viết tới mình sẽ cùng tìm hiểu về HOT Observable và COLD Observable xem nó là gì và tác dụng từng loại nhé. Happy coding !!!

Reference

• Making threading easy in Android with RxJava