Giải thích bằng một hình ảnh về điều khiển luồng trong RxJava

2016-04-24

Vào thứ Sáu tuần trướcsv 88, tôi đã thảo luận với đội nhóm RxJava Cách sử dụng và cơ chế thực hiện của nó. Trong cuộc thảo luậnsv 88, thầy Giang Giang đã đặt ra một câu hỏi thú vị: Nếu trong chuỗi gọi có nhiều hơn một lệnh subscribeOn và observeOn, thì điều đó sẽ dẫn đến kết quả gì? Trong trường hợp này, khi bạn sử dụng nhiều lệnh subscribeOn và observeOn liên tiếp, luồng dữ liệu sẽ thay đổi dựa trên lịch trình mà bạn thiết lập. Mỗi lệnh subscribeOn sẽ xác định luồng mới cho việc đăng ký nguồn dữ liệu, trong khi observeOn lại quyết định nơi mà các hoạt động quan sát sẽ diễn ra. Điều này có thể gây ra sự chồng chéo hoặc xung đột nếu không được quản lý cẩn thận, dẫn đến việc khó kiểm soát dòng chảy của dữ liệu. Thầy Giang Giang nhấn mạnh rằng, để tránh những vấn đề không mong muốn, cần phải hiểu rõ thứ tự và vị trí của từng lệnh trong chuỗi gọi. Việc hiểu rõ cách thức hoạt động của chúng không chỉ giúp tối ưu hóa hiệu suất mà còn đảm bảo tính ổn định cho toàn bộ ứng dụng.

Đây thực sự là một vấn đề quan trọngsv 88, bởi vì trong mọi tình huống, chúng ta cần phải hiểu rõ từng dòng mã mà mình viết đang chạy trên luồng nào. Điều này không được để mơ hồ chút nào. Việc xác định chính xác luồng thực thi của mã giúp đảm bảo tính ổn định và hiệu suất của ứng dụng, đồng thời giảm thiểu các rủi ro về lỗi đồng bộ hóa hoặc xung đột tài nguyên. Chúng ta không thể bỏ qua yếu tố này nếu muốn xây dựng phần mềm chất lượng cao và đáng tin cậy.

Giả sử có đoạn mã giả dưới đây:

							
								

									

										

											

												
1
2
3
4
5
6
7
8
9
10
11
12
13

											
											

												
													
														Observable
													.
													create
													(...)
													
	.
													lift1
													(...)
													
	.
													subscribeOn
													(
													scheduler1
													)
													
	.
													lift2
													(...)
													
	.
													observeOn
													(
													scheduler2
													)
													
	.
													lift3
													(...)
													
	.
													subscribeOn
													(
													scheduler3
													)
													
	.
													lift4
													(...)
													
	.
													observeOn
													(
													scheduler4
													)
													
	.
													doOnSubscribe
													(...)
													
	.
													subscribeOn
													(
													scheduler5
													)
													
	.
													observeOn
													(
													scheduler6
													)
													
	.
													subscribe
													(...);
													

											
										
									
								
							

Trong đókeo nha cai hom nay, lift1, lift2, lift3, lift4 đề cập đến các thao tác biến đổi được thực hiện dựa trên lift, chẳng hạn như filter, map, reduce và nhiều chức năng khác. Những công cụ này đóng vai trò quan trọng trong việc xử lý dữ liệu một cách hiệu quả và linh hoạt.

Trong đoạn mã này:

• Bạn có thể cho biết mã nguồn được gán cho lift1keo nha cai hom nay, lift2, lift3 và lift4 sẽ được thực thi trên luồng nào? Có thể các đoạn mã này đã được thiết kế để chạy trên những luồng khác nhau nhằm tối ưu hóa hiệu suất. Nhưng cụ thể từng mã nguồn sẽ chạy trên luồng nào, ta cần xem xét kỹ cách thức triển khai của lập trình viên. Có khả năng mỗi lift được gán riêng một luồng để xử lý yêu cầu nhanh hơn, hoặc tất cả chúng có thể cùng chia sẻ một luồng nhưng có cơ chế quản lý riêng biệt.
• Mã được chỉ định bởi doOnSubscribe sẽ được thực thi trên luồng nào?
• Mã tạo ra sự kiện (mã được chỉ định bởi create) sẽ được thực thi trên luồng nào?
• Mã tiêu thụ sự kiện (mã được chỉ định bởi subscribe) sẽ được thực thi trên luồng nào?

Rất có thể nhiều bạn sẽ cảm thấy đoạn mã này hơi khiến đầu óc quay cuồng và dường như không thực tế chút nào. Đúng vậysv 88, trong thực tế, mã nguồn thường không phức tạp đến mức đó, nhưng việc hiểu được nó sẽ giúp chúng ta nắm bắt toàn bộ quy trình hoạt động của RxJava một cách rõ ràng hơn. Một khi đã hiểu được nguyên lý này, bạn sẽ nhận ra rằng những gì mà thư viện này làm không quá xa vời so với những gì bạn tưởng tượng ban đầu.

Hình ảnh phía trên minh họa quá trình truyền dòng điều khiển trong một chuỗi gọi điển hình của RxJava. Quy trình này có thể được chia thành hai giai đoạn chính: --- **Giai đoạn 1:** Tại giai đoạn đầu tiênsv 88, nguồn dữ liệu (source) sẽ tạo ra các sự kiện hoặc dữ liệu và gửi chúng qua chuỗi xử lý. Mỗi đối tượng trong chuỗi, như các nhà quan sát (observer) hoặc bộ xử lý (operator), sẽ thực hiện nhiệm vụ riêng biệt để biến đổi hoặc lọc dữ liệu trước khi chuyển tiếp nó sang bước tiếp theo. Điều này giống như một dòng thác mà mỗi tảng đá cản lại một chút trước khi nước tiếp tục chảy xuống. --- **Giai đoạn 2:** Ở giai đoạn thứ hai, dữ liệu đã qua xử lý sẽ được giao cho người nhận cuối cùng (subscriber). Người nhận này sẽ thực hiện các hành động dựa trên dữ liệu nhận được, chẳng hạn như hiển thị kết quả, lưu trữ thông tin hoặc thực hiện các thao tác khác. Đây là lúc toàn bộ chuỗi hoạt động được hoàn thiện và giá trị cuối cùng được sử dụng đúng mục đích. --- Cấu trúc này không chỉ giúp quản lý dòng dữ liệu một cách hiệu quả mà còn tăng tính linh hoạt trong việc xử lý các yêu cầu phức tạp.

1. Giai đoạn kích hoạt. Toàn bộ dòng chảy sự kiện bất đồng bộ được khởi xướng bởi hà Nó tạo ra một quá trình kích hoạt ngược (từ dưới lên trên)keo 88, đi qua từng Observable và OnSubscribe trung gian, cho đến khi đạt được Observable đầu tiên (nguồn phát sinh các sự kiện). Điều này tương ứng với các bước (1) và (2) trong hình. Thông thường, giai đoạn này kết thúc sau khi thực hiện một lần từ dưới lên trên. Tuy nhiên, trong một số trường hợp đặc biệt, quá trình kích hoạt có thể còn bao gồm thêm các bước bổ sung để đảm bảo rằng mọi Observable đều đã sẵn sàng nhận dữ liệu. Điều này giúp tăng cường độ ổn định và tính nhất quán trong dòng chảy sự kiện, đảm bảo rằng không có Observable nào bị bỏ sót hoặc gặp vấn đề trong quá trình xử lý.
2. Giai đoạn khởi phát sự kiện. Observable đầu tiên bắt đầu tạo ra các sự kiệnkeo 88, và dòng sự kiện sẽ bắt đầu di chuyển theo hướng từ trên xuống dưới, đi qua từng Observable trung gian cho đến khi đến Subscriber (người tiêu thụ sự kiện). Điều này tương ứng với phần (3) trong hình. Khác với giai đoạn trước, sự kiện không chỉ đơn giản truyền từ nguồn lên người dùng một lần mà nó là dòng sự kiện liên tục, bao gồm nhiều sự kiện khác nhau kết nối với nhau. Dòng sự kiện này mang ý nghĩa quan trọng trong việc xử lý luồng dữ liệu, giúp các nhà phát triển có thể quản lý và xử lý dữ liệu một cách linh hoạt. Mỗi Observable trung gian có thể thực hiện các thao tác như lọc, biến đổi hoặc thậm chí kết hợp dòng sự kiện trước khi nó được Subscriber nhận và xử lý cuối cùng. Đây chính là yếu tố cốt lõi của lập trình phản ứng (reactive programming), nơi mà các sự kiện được xử lý một cách tuần tự và hiệu quả.

Chúng ta cùng phân tích toàn bộ quy trình nàysv 88, trong đó có một số điểm cần được giải thích rõ hơn (lưu ý: quá trình phân tích này đề cập đến một số chi tiết thực hiện của RxJava. Nếu bạn không quan tâm đến các chi tiết kỹ thuật, có thể bỏ qua đoạn này và chuyển thẳng sang phần kết luận ở phía sau).

• Trong hình ảnhsv 88, mục (1) đại diện cho việc gọi phương thứ call của Observable cấp trên, một phương thức không có giá trị trả về. Do đó, nó có khả năng chuyển đổi luồng thực thi sang một luồng khác, biến tiến trình thành bất đồng bộ. Chính vì lý do này, nó được biểu thị bằng đường viền đứt để chỉ ra sự thay đổi trạng thái và khả năng chạ
• Trong hìnhkeo nha cai hom nay, (2) đại diện cho thao tác lift được chỉ định bởi phương thứ call, đây là một phương thức có giá trị trả về (nhận vào một Subscriber và trả về một Subscriber mới). Do đó, nó chỉ có thể được gọi đồng bộ và không thể chuyển đổi luồng. Vì lý do này, nó được biểu thị bằng đường nét thực trong hình.
• Trong hình ảnhkeo nha cai hom nay, mục (3) ám chỉ việc gọi Subscriber tương ứng với Observable cấp dưới (bao gồm các phương thức onNext, onCompleted và onError), tất cả đều là các phương thức không có giá trị trả về. Do đó, chúng có thể thay đổi luồng thực thi sang một luồng khác, biến hoạt động thành bất đồng bộ. Chính vì lý do này, các mối liên kết ở đây được biểu thị bằng đường viền đứt để nhấn mạnh tính chất linh hoạt và tách biệt của chúng.
• Hàm observeOn được xây dựng dựa trên lift và hành động chuyển đổi luồng sẽ xảy ra tại Subscriber (onNextkeo 88, onCompleted, onError). Do đó, nó có tác động đến tất cả các thay đổi của lift nằm phía dưới nó trong quy trình (3), ảnh hưởng trực tiếp đến cách mà dữ liệu được xử lý sau này.
• Hàm subscribeOn không được xây dựng dựa trên liftkeo 88, mà thay vào đó nó thực hiện việc chuyển đổi luồng ngay trước khi gọi OnSubscribe của Observable cấp cao hơn. Điều này có nghĩa là nó sẽ ảnh hưởng đến (1) toàn bộ chuỗi các cuộc gọi OnSubscribe nằm phía trên trong dòng chảy, từ thời điểm đó cho đến khi nguồn sự kiện được tạo ra. Sau đó, (3) tất cả các hoạt động lift trong dòng chảy cũng sẽ tiếp tục chạy trên luồng mới đã được chuyển đổi, trừ khi chúng gặp một lệ Khi đó, luồng thực thi sẽ được điều chỉnh lại theo cách đã định nghĩa trong hà
• doOnSubscribe có một đặc điểm hơi khác biệt. Mặc dù nó cũng được xây dựng dựa trên cơ chế liftsv 88, nhưng đoạn mã mà nó chỉ định sẽ được thực hiện trong phương thức call của Operator, không giống như các thao tác lift khác, nơi mã được thự Do đó, luồng thực thi của doOnSubscribe sẽ chịu ảnh hưởng từ phương thức subscribeOn phía dưới của nó. Điều này làm cho nó trở thành một phần quan trọng trong việc kiểm soát dòng dữ liệu và đảm bảo sự đồng bộ giữa các giai đoạn trong chuỗi xử lý.

Kết hợp với phân tích phía trênsv 88, chúng ta đi theo hướng mà mũi tên trong biểu đồ quy trình trước đó chỉ dẫn:

• Bắt đầu bằng cách gọi phương thức subscribekeo 88, bạn sẽ đi theo lộ trình được chỉ ra trong sơ đồ dòng chảy trước đây (1)->(2)->(1)->(2)...->(1) (tức là giai đoạn kích hoạt), di chuyển từ dưới lên trên. Khi đi qua mỗi lệnh subscribeOn, luồng thực thi sẽ thay đổi một lần; mỗi sự thay đổi này sẽ ảnh hưởng đến mã được chỉ định bởi doOnSubscribe và mã tạo ra các sự kiện (được chỉ định bởi create) nằm phía sau nó trên cùng một đường dẫn (tức là phía bên trên).
• Sau khi đến nguồn của sự kiện (mã được chỉ định bởi create)keo 88, nó chuyển sang giai đoạn phát tán sự kiện.
• Tiếp theokeo 88, đi dọc theo lộ trình (3) (tức là giai đoạn phát tán sự kiện), từ phía nguồn đến phía cuối dòng chảy, mỗi khi gặp một `observeOn`, luồng thực thi sẽ được chuyển đổi một lần. Sự thay đổi này không chỉ tác động lên môi trường thực thi của chính nó mà còn ảnh hưởng đến tất cả các hoạt động `lift` phía sau (tức là ở hướng dòng chảy xuống) trên cùng một đường dẫn, cho đến khi mã xử lý sự kiện được tiêu thụ (được xác định bởi mã `subscribe`). Quá trình này giống như một chuỗi phản ứng dây chuyền, nơi mỗi bước chuyển tiếp đều thiết lập bối cảnh mới cho các bước tiếp theo trong dòng sự kiện.

Bây giờkeo 88, nếu mô tả lại những điều này theo cách khác, thì rất dễ dàng để rút ra các kết luận sau:

• Code được chỉ định bởi doOnSubscribe và code tạo ra sự kiện (code được chỉ định bởi create) sẽ được thực hiện trên Scheduler gần nhất mà subscribeOn ở phía dưới chúng. Nếu không có subscribeOn nào nằm phía dưới chúngkeo 88, chúng sẽ chạy trên chính thread mà phương thức subscribe được gọi. Lưu ý rằng đây là thread mà phương thức subscribe được gọi, không phải thread nơi code của subscribe được thực thi, hai điều này hoàn toàn khác nhau. Trong trường hợp không có subscribeOn nào nằm giữa doOnSubscribe hoặc create và điểm gọi subscribe(), các hoạt động này sẽ diễn ra hoàn toàn song song với luồng chính mà subscribe() được khởi chạy. Điều này đặc biệt quan trọng khi bạn cần hiểu rõ luồng dữ liệu và hành vi đa luồng trong các ứng dụng sử dụng RxJava. Hiểu rõ điểm khác biệt giữa luồng gọi subscribe và luồng thực thi của subscribe giúp lập trình viên kiểm soát hiệu quả hơn các tiến trình đồng thời trong ứng dụng của mình.
• Các hoạt động lift thông thường (như filterkeo nha cai hom nay, map, reduce, v.v.) và mã xử lý sự kiện (code được chỉ định trong subscribe) sẽ thực hiện trên Scheduler gần nhất mà observeOn được gọi ở phía trên của chúng. Nếu không tìm thấy bất kỳ observeOn nào phía trên, chúng sẽ chạy trên Scheduler đầu tiên được chỉ định bởi subscribeOn ở phần cao nhất của chuỗi gọi. Và nếu không có subscribeOn nào được gọi, thì chúng sẽ thực hiện trực tiếp trên thread mà phương thức subscribe được gọi. Trong trường hợp này, mỗi bước trong chuỗi quan trọng như observeOn hay subscribeOn đều đóng vai trò xác định nơi mà luồng dữ liệu sẽ chạy. Điều này giúp lập trình viên có thể kiểm soát được luồng thực thi của ứng dụng, đặc biệt là khi làm việc với các tác vụ đồng bộ hoặc bất đồng bộ phức tạp. Với khả năng linh hoạt này, RxJava cho phép bạn tối ưu hiệu suất và tăng cường khả năng mở rộng của ứng dụng một cách hiệu quả.

Áp dụng những kết luận này vào đoạn mã ban đầu ở phần mở đầu bài viếtkeo 88, chúng ta nhanh chóng nhận được:

• Mã tạo ra sự kiện (mã được chỉ định bởi create) sẽ được thực thi trên scheduler1;
• Mã được chỉ định bởi lift1 và lift2 sẽ được thực thi trên scheduler1;
• Mã được chỉ định bởi lift3 và lift4 sẽ được thực thi trên scheduler2;
• Mã được chỉ định bởi doOnSubscribe sẽ được thực thi trên scheduler5;
• Mã tiêu thụ sự kiện (mã được chỉ định bởi subscribe) sẽ được thực thi trên scheduler6.