Giải thích bằng một bức ảnh về kiểm soát luồng trong RxJava

2016-04-24

Vào thứ Sáu tuần trướcxóc đĩa, tôi đã thảo luận với đội ngũ của mình về vấn đề này RxJava Cách sử dụng và cơ chế thực hiện của điều đó. Trong cuộc thảo luậnxóc đĩa, 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 gì sẽ xảy ra?

Đây thực sự là một vấn đề rất quan trọngxóc đĩa, 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 được thực thi trên luồng nào. Điều này hoàn toàn không thể mơ hồ được. Việc xác định chính xác luồng thực hiện giúp đảm bảo rằng ứng dụng hoạt động ổn định và tránh được các lỗi khó phát hiện liên quan đến đồng bộ hóa hoặc xung đột tài nguyên giữa các luồng. Chính vì vậy, việc theo dõi và kiểm soát luồng thực thi của mã nguồn luôn là ưu tiên hàng đầu trong quá trình phát triển phần mềm phức tạp.

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 đótỷ số bóng đá hôm 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 (lọc), map (duyệt), reduce (tích lũy). Bên cạnh đó, chúng còn bao gồm nhiều kỹ thuật khác giúp xử lý dữ liệu một cách linh hoạt và hiệu quả trong lập trình.

Trong đoạn mã này:

• Bạn có thể cho biết mã của lift1tỷ số bóng đá hôm nay, lift2, lift3 và lift4 sẽ được thực hiện trong luồng nào? Điều này phụ thuộc vào cách mà chương trình của bạn đã được cấu trúc và cách các luồng được quản lý bên trong hệ thống. Có khả năng rằng mỗi lift sẽ được gán cho một luồng riêng biệt để xử lý đồng thời, nhưng điều này cũng có thể thay đổi tùy thuộc vào logic cụ thể của ứng dụng. Bạn có thể kiểm tra lại mã nguồn để hiểu rõ hơn về cách thức hoạt động của từng phần tử này.
• Mã được chỉ định bởi doOnSubscribe sẽ được thực thi trên luồng nào?
• Mã tạo ra các 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ụ các 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 khá rối rắm và khó hiểulịch bóng đá trực tiếp, thực tế thì nó không thường xuất hiện trong công việc hàng ngày. Thực chất, phần lớn mã nguồn thực tế không phức tạp đến vậy, nhưng việc hiểu được nó sẽ giúp chúng ta nắm rõ toàn bộ quy trình hoạt động của RxJava.

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. Quá trình này có thể được chia thành hai giai đoạn chính: --- ### Giai đoạn 1: Khởi tạo và kích hoạt dòng dữ liệu Giai đoạn đầu tiên là khi nguồn dữ liệu (source) được khởi tạo và bắt đầu đẩy các sự kiện đến chuỗi xử lý. Điều này thường được thực hiện bằng cách sử dụng các phương thức như ` create()` hoặc các hàm tiện ích khác để tạo ra luồng dữ liệu. Tại thời điểm nàyxóc đĩa, các nhà quan sát (observer) sẽ chờ nhận các sự kiện từ nguồn và sẵn sàng xử lý chúng. --- ### Giai đoạn 2: Xử lý và phản hồi dữ liệu Giai đoạn thứ hai tập trung vào việc xử lý dữ liệu sau khi nó đã được nhận từ nguồn. Trong giai đoạn này, các nhà vận hành (operators) như `map()`, `filter()`, hoặc `flatMap()` sẽ được sử dụng để biến đổi, lọc hoặc kết hợp dữ liệu trước khi chuyển tiếp cho nhà quan sát cuối cùng. Sau khi hoàn tất xử lý, nhà quan sát sẽ nhận dữ liệu và đưa ra phản hồi. --- Quá trình này giúp quản lý dòng chảy dữ liệu một cách linh hoạt và hiệu quả, đồng thời đảm bảo rằng các tác vụ phức tạp có thể được thực hiện tuần tự hoặc song song tùy thuộc vào yêu cầu.

1. Giai đoạn khởi động. Toàn bộ dòng chảy sự kiện bất đồng bộ được kích hoạt bởi lệ Điều này khởi xướng một quá trình ngược dòng (từ hạ nguồn đến thượng nguồn)lịch bóng đá trực tiếp, đi qua từng Observable và OnSubscribe ở giữa, 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 chỉ kết thúc sau khi thực hiện một lần gọi từ hạ nguồn lên thượng nguồn. Tuy nhiên, trong một số trường hợp đặc biệt, quá trình này có thể trở nên phức tạp hơn khi có nhiều chuỗi phụ hoặc các Observable bổ sung xuất hiện. Điều này dẫn đến việc có thêm các luồng dữ liệu bổ sung, làm tăng tính linh hoạt nhưng cũng cần phải quản lý cẩn thận để tránh xảy ra lỗi hoặc tình trạng không mong muốn trong hệ thống.
2. Giai đoạn phát tán sự kiện. Quá trình bắt đầu khi Observable đầu tiên bắt đầu tạo ra các sự kiệntỷ số bóng đá hôm nay, sau đó dòng sự kiện sẽ di chuyển theo hướng từ trên xuống dưới qua từng Observable trung gian, cuối cùng đế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 biệt so với giai đoạn trước đó, dòng sự kiện không chỉ đơn thuần là một lần truyền từ nguồn lên đích mà còn bao gồm chuỗi nhiều sự kiện liên tiếp, tạo thành một dòng chảy mạnh mẽ và liên tục. Dòng sự kiện này không chỉ đơn giản là một tín hiệu lẻ tẻ mà nó mang tính chất kết nối và tuần tự, giúp cho việc xử lý dữ liệu trở nên linh hoạt hơn. Mỗi Observable đóng vai trò như một mắt xích quan trọng trong chuỗi, đảm bảo rằng các sự kiện được chuyển tiếp một cách chính xác và có thứ tự đến Subscriber cuối cùng.

Chúng ta cùng phân tích toàn bộ quy trình nàyxóc đĩa, 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 ở đây liên quan đến một số chi tiết kỹ thuật của RxJava, nếu bạn không quan tâm đến những chi tiết này, có thể bỏ qua đoạn này và trực tiếp xem phần kết luận phía sau).

• Trong hìnhlịch bóng đá trực tiếp, (1) ám chỉ việc gọi phương thứ call của Observable cấp trên, đây là một phương thức không trả về giá trị, do đó nó có thể chuyển đổi luồng thực thi sang một luồng khác, từ đó 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.
• Trong hình ảnhtỷ số bóng đá hôm nay, mục (2) ám chỉ đến phương thứ call được chỉ định bởi lift, đây là một phương pháp 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ộ mà 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 ảnhxóc đĩa, (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ể chuyển đổi luồng thực thi sang một luồng khác, biến quy trình thành bất đồng bộ. Chính vì lý do này, mối liên kết ở đây được biểu thị bằng đường viền đứt.
• Phương thức observeOn được xây dựng dựa trên lift và hoạt động chuyển đổi luồng thực hiện tại Subscriber (onNextlịch bóng đá trực tiếp, onCompleted, onError). Do đó, nó ảnh hưởng đến toàn bộ các thay đổi của lift ở tất cả các bước phía dưới trong chuỗi xử lý. Điều này có nghĩa là bất kỳ sự điều chỉnh nào mà observeOn thực hiện sẽ lan tỏa và tác động trực tiếp lên tất cả các bước tiếp theo, tạo ra một chuỗi phụ thuộc chặt chẽ trong luồng dữ liệu.
• Mặc dù subscribeOn không được xây dựng dựa trên lifttỷ số bóng đá hôm nay, nó thay đổi luồng thực thi bằng cách chuyển đổi thread ngay khi gọi OnSubscribe của Observable cấp trên. Do đó, nó tác động (1) lên toàn bộ chuỗi các OnSubscribe nằm phía trên trong dòng chảy, kéo dài từ điểm chuyển đổi cho đến khi nguồn sự kiện được tạo ra; tiếp theo, (3) tất cả các thao tác lift trong dòng chảy cũng sẽ chạy trên thread mới này, trừ khi gặp một lệnh observeOn, nơi nó sẽ thiết lập lại luồng thực thi sang thread khác.
• doOnSubscribe có một chút khác biệt so với các phương thức lift khác. Mặc dù nó cũng được thực hiện dựa trên cơ chế của liftlịch bóng đá trực tiếp, nhưng đoạn mã mà nó chỉ định sẽ chạy trong phương thứ call, không phải trong Subscriber như những trường hợp lift thông thường. Chính vì điều này, luồng thực thi của doOnSubscribe sẽ bị ảnh hưởng bởi phương thức subscribeOn phía dưới nó. Điều đó có nghĩa là nó sẽ chạy trên dòng thread được chỉ định bởi subscribeOn, thay vì thread mà nó được tạo ra ban đầu.

Kết hợp với phân tích trêntỷ số bóng đá hôm nay, chúng ta sẽ đi theo hướng mà mũi tên chỉ trong lưu đồ quy trình trước đó:

• Bắt đầu từ việc gọi phương thức subscribexóc đĩa, đi theo con đường (1)->(2)->(1)->(2)…->(1) trong sơ đồ lưu đồ trước đây (tức là giai đoạn kích hoạt), di chuyển từ phía dưới lên phía trên. Mỗi khi đi qua một subscribeOn, luồng thực thi sẽ thay đổi một lần; sự thay đổi này sẽ ảnh hưởng đến mã được chỉ định trong doOnSubscribe và mã tạo ra các sự kiện (được chỉ định bởi create) ở phía sau (tức là phía upstream) trên cùng một nhánh đó.
• Sau khi đến nguồn của các sự kiện (mã được chỉ định bởi create)lịch bóng đá trực tiếp, chúng ta tiến vào giai đoạn phát tán sự kiện.
• Tiếp theotỷ số bóng đá hôm nay, đi dọc theo lộ trình (3) (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 observOn, luồng thực thi sẽ được chuyển đổi một lần; mỗi lần thay đổi này sẽ ảnh hưởng đến toàn bộ các hoạt động lift phía dưới trong cùng nhánh này, cho đến khi mã xử lý sự kiện được thực hiện (mã được chỉ định bởi subscribe). Những sự thay đổi này không chỉ ảnh hưởng đến cách các sự kiện được xử lý mà còn định hình bối cảnh thực thi cho từng bước tiếp theo trong chuỗi này.

Bây giờtỷ số bóng đá hôm nay, nếu chúng ta đổi cách diễn đạt trước đó, kết luận sau đây rất dễ dàng được rút ra:

• 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 thi trên Scheduler gần nhất mà subscribeOn chỉ định ở phía dưới của dòng xử lý; nếu không có subscribeOn nào nằm phía dướitỷ số bóng đá hôm nay, chúng sẽ chạy trên chính thread mà phương thức subscribe được gọi. Hãy lưu ý rằng đây là thread mà phương thức subscribe được thực hiện, 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.
• Các hoạt động lift thông thường (như filterxóc đĩa, map, reduce,...) và mã xử lý sự kiện (code được chỉ định bởi subscribe) sẽ thực hiện trên Scheduler gần nhất được chỉ định bởi observeOn ngay phía trên chúng trong chuỗi gọi; 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 tìm thấy, chúng sẽ chạy trên chính luồng mà phương thức subscribe được gọi. Trong trường hợp không có gì khác được cấu hình, điều này đảm bảo rằng luồng thực thi luôn được quản lý một cách logic và dễ kiểm soát. Điều này đặc biệt hữu ích khi bạn cần đồng bộ hóa hoặc phân tán công việc giữa các luồng khác nhau để tránh làm nghẽn luồng chính hoặc cải thiện hiệu suất tổng thể.

Áp dụng những kết luận này lên đoạn mã ban đầu trong bài viết nàyxóc đĩa, chúng ta nhanh chóng nhận được:

• Mã tạo ra các 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ụ các sự kiện (mã được chỉ định bởi subscribe) sẽ được thực thi trên scheduler6.