Reactive Programming Là Gì

Reactive programing là khái niệm khá trừu tượng và khó tiếp cận với người mới bắt đầu, chuẩn bị tinh thần đọc bài này vài lần trong vài ngày thì mới mong thẩm thấu hết.

Bạn đang xem: Reactive programming là gì


Reactive programing là gì?

Reactive programming is programming with asynchronous data streams

Tạm dịch: Reactive programming là lập trình xử lý với dữ liệu không tuần tự (async) như stream

Có khái niệm mới stream


*
*

10 năm trước, mọi việc chỉ đơn giản là submit toàn bộ giá trị các field lên backend xử lý, rồi đơn thuần hiển thị kết quả trả về, bây giờ user thích real-time feedback, bấm “like” một phát là đầu bên kia thấy được liền.

Những event real-time như thế, user khoái, chúng ta cần có một công cụ lập trình để làm việc đó, Reactive Program ra đời cũng từ yêu cầu của user.

Implement hộp thoại “Who to follow” của twitter

Mình sẽ sử dụng RxJS trong ví dụ, vì mình chỉ biết javascript thôi các bạn.

*

Tính năng chính của hộp thoại này

Vừa mở lên, load data từ API, hiển thị 3 tài khoảnClick “Refresh”, hiển thị 3 tài khoản khácKhi click “x”, xóa tài khoản đó khỏi danh sách, hiển thị một tài khoản khác.

Chúng ta tiếp cận với vấn đề này như thế nào, gần như mọi thứ có thể xem là stream.

Xem thêm: Hướng Dẫn Thay Đổi Icon Start Win 7 Toàn Tập, Cách Đổi Nút Start Trên Windows 10

Load dữ liệu lúc đầu

Bắt đầu với tính năng đơn giản nhất “Mới vào, load 3 account từ API”. (1) gửi 1 request (2) nhận response (3) render kết quả

Lúc bắt đầu chúng ta chỉ có 1 request, mọi thứ rất đơn giản, yên tâm là nó sẽ phức tạp dần lên khi có nhiều request. Mô phỏng nó như data stream, stream này chỉ có 1 emit value.

——a——-|—>Khi có một event request xảy ra, nó báo 2 việc: khi nào và cái gì. Khi nào event này được emit và cái gì chính là value được emit (url string)

Trong Rx, bà con gọi stream là Observable, mình thích gọi là stream hơn

var requestStream = Rx.Observable.just("https://api.github.com/users");Khi emit value, chúng ta subscribe để thực thi một hành động tiếp theo

requestStream.subscribe( requestUrl => {// execute the request jQuery.getJSON(requestUrl, function(responseData) { // ... });}Cái response của request cũng là một dạng stream, dữ liệu sẽ đến tại một thời điểm không xác định trong tương lai

requestStream.subscribe(function(requestUrl) { // execute the request var responseStream = Rx.Observable.create(function (observer) { jQuery.getJSON(requestUrl) .done(function(response) { observer.onNext(response); }) .fail(function(jqXHR, status, error) { observer.onError(error); }) .always(function() { observer.onCompleted(); }); }); responseStream.subscribe(function(response) { // do something with the response });}Rx.Observable.create() sẽ tạo ra những stream mới, qua việc thông báo cho các observer đang subscriber các sự kiện onNext(), onError().

Nó giống cách chạy của Promise lắm đúng không? Vâng Observable là một dạng Promise++, phiên bản mở rộng.

Chúng ta có 1 subscribe bên trong 1 subscribe khác, nó giống như callback hell. Thêm nữa việc tạo responseStream hoàn toàn độc lập với requestStream. Trong Rx chúng ta có một cách đơn giản để transform và tạo một stream mới từ những thằng khác

Hàm map(f), sẽ lấy từng giá trị của stream A, gọi function f(), và trả về giá trị cho stream B. Tạo một stream này từ stream khác, y như hàm map của array thôi mà.

var responseMetastream = requestStream .map(function(requestUrl) { return Rx.Observable.fromPromise(jQuery.getJSON(requestUrl)); });Sau đó chúng ta tạo một stream của stream metastream. Bắt đầu phức tạp rồi đó. Metastream là 1 stream mà mỗi cái value được emit sẽ trỏ ra 1 stream khác. Trong ví dụ, mỗi URL request, được trỏ đến một stream promise chứa response

*

Với responseStream, chúng ta chỉ một đơn giản một stream chứa response, nên việc tạo một metastream cho response sẽ rối và không cần. Mỗi giá trị được emit của response sẽ là một object JSON, không phải một Promise của object JSON. Sử dụng .flatMap() để gộp tất cả response thành 1 stream, .flatMap là operator để xử lý dữ liệu async trong Rx

var responseStream = requestStream .flatMap(function(requestUrl) { return Rx.Observable.fromPromise(jQuery.getJSON(requestUrl)); });

*

responseStream được khai báo bởi requestStream, nếu sau này có thêm các sự kiện trên requestStream, chúng ta sẽ có một event response tương ứng trên responseStream

requestStream: --a-----b--c------------|->responseStream: -----A--------B-----C---|->Sau khi có được responseStream, chúng ta render thôi

responseStream.subscribe(function(response) { // render `response` to the DOM however you wish});Toàn bộ bode bây giờ

var requestStream = Rx.Observable.just("https://api.github.com/users");var responseStream = requestStream .flatMap(function(requestUrl) { return Rx.Observable.fromPromise(jQuery.getJSON(requestUrl)); });responseStream.subscribe(function(response) { // render `response` to the DOM however you wish});

Nút refresh

JSON trả về từ API sẽ có 100 user, nó chỉ cho thêm offset, không cho set page size, chúng ta chỉ cần 3 user, lãng phí hết 97 user. Tạm thời không quan tâm phần này, chúng ta sẽ cache lại cái response sau.

Khi click nút refresh, requestStream sẽ emit một URL mới, sau đó chúng ta nhận được một response mới. Chúng ta cần 2 thứ:

1 stream cho sự kiện click -> refreshStreamcập nhập lại requestStream để nó phụ thuộc vào refreshStream

RxJS có hàm để chuyển event thành stream

var refreshButton = document.querySelector(".refresh");var refreshClickStream = Rx.Observable.fromEvent(refreshButton, "click");Click refresh nó không có URL kèm theo, chúng ta phải nhét cái URL bằng code. Map vào URL với giá trị offset ngẫu nhiên

var requestStream = refreshClickStream .map(function() { var randomOffset = Math.floor(Math.random()*500); return "https://api.github.com/users?since=" + randomOffset; });Tới đây, chắc chắn mở app lên không thấy gì cả, không có request nào được gửi đi, chỉ click refresh thì mới thấy.

Phải tách stream này ra riêng

var requestOnRefreshStream = refreshClickStream .map(function() { var randomOffset = Math.floor(Math.random()*500); return "https://api.github.com/users?since=" + randomOffset; }); var startupRequestStream = Rx.Observable.just("https://api.github.com/users");Sau đó mới .merge() lại

stream A: ---a--------e-----o----->stream B: -----B---C-----D--------> vvvvvvvvv merge vvvvvvvvv ---a-B---C--e--D--o----->var requestOnRefreshStream = refreshClickStream .map(function() { var randomOffset = Math.floor(Math.random()*500); return "https://api.github.com/users?since=" + randomOffset; }); var startupRequestStream = Rx.Observable.just("https://api.github.com/users");var requestStream = Rx.Observable.merge( requestOnRefreshStream, startupRequestStream);Có cách gọn hơn, không cần đến một stream trung gian

var requestStream = refreshClickStream .map(function() { var randomOffset = Math.floor(Math.random()*500); return "https://api.github.com/users?since=" + randomOffset; }) .merge(Rx.Observable.just("https://api.github.com/users"));Thậm chí gọn hơn nữa

var requestStream = refreshClickStream .map(function() { var randomOffset = Math.floor(Math.random()*500); return "https://api.github.com/users?since=" + randomOffset; }) .startWith("https://api.github.com/users");Chủ ý nãy giờ là giải thích .startWith() đó. Tuy nhiên là còn có thể tốt hơn nếu chúng ta không lặp lại URL. Làm việc đó bằng cách dời thằng startWith() ngay sau refreshClickStream, để giả lập sự kiện refresh khi vừa mới mở

var requestStream = refreshClickStream.startWith("startup click") .map(function() { var randomOffset = Math.floor(Math.random()*500); return "https://api.github.com/users?since=" + randomOffset; });Khi click nút refresh, chúng ta cũng sẽ remove 3 thằng user đang hiển thị, như vậy chúng ta sẽ subscribe trên refreshClickStream

refreshClickStream.subscribe(() => { // clear 3 sugesstion})Tuy nhiên, responseStream cũng đang có 1 subscribe ảnh hướng đến việc render, như vậy việc render này cũng tạo thêm 1 stream (có 2 sự kiện emit value để render)

var suggestion1Stream = responseStream .map(function(listUsers) { // get one random user from the list return listUsers; });Chúng ta cũng sẽ có suggestion2Stream, suggestion3Stream, suggestionNStream hoàn toàn giống với suggestion1Stream, nhưng mình sẽ để các bạn tự suy nghĩ cách giải quyết. Ví dụ này chỉ đề cập đến suggestion1Stream

Thay vì render trên subscribe của responseStream

suggestion1Stream.subscribe(function(suggestion) { // render the 1st suggestion to the DOM});Quay lại vấn đề “click refresh, xóa suggestion”, chúng ta đưa vào sugesstion1Stream giá trị null khi refresh

var suggestion1Stream = responseStream .map(function(listUsers) { // get one random user from the list return listUsers; }) .merge( refreshClickStream.map(function(){ return null; }) );Với trường hợp null, đơn giản render thông báo

suggestion1Stream.subscribe(function(suggestion) { if (suggestion === null) { // hide the first suggestion DOM element } else { // show the first suggestion DOM element // and render the data }});Hình dung quá trình này như sau, trong đó N là giá trị null