Xin chào các bạn, bài viết hôm nay mình sẻ giới thiệu tiếp về thư viện System.Reactive (Rx.NET) của Microsoft nó hay ho ở những điểm gì.

[C#] Reactive (Rx.NET) là gì, và một số ví dụ về sử dụng Rx

System.Reactive có thể được sử dụng để giải quyết nhiều vấn đề khác nhau trong lập trình. Dưới đây là một số ứng dụng hay nhất của System.Reactive trong C#:

1. Xử lý sự kiện: System.Reactive cho phép dễ dàng xử lý sự kiện từ các nguồn khác nhau, chẳng hạn như các thành phần giao diện người dùng, các tập tin log hoặc dữ liệu đến từ mạng.

2. Tương tác với dữ liệu bất đồng bộ: Khi tương tác với dữ liệu bất đồng bộ, System.Reactive cung cấp một cách để quản lý các giá trị dữ liệu đến và đi một cách an toàn và dễ dàng.

3. Lập trình hướng reative: System.Reactive cung cấp một cách tiếp cận mới trong lập trình hướng reative, cho phép các lập trình viên xây dựng các ứng dụng dựa trên những thay đổi trong các luồng dữ liệu.

4. Đồng bộ hóa các sự kiện không đồng bộ: System.Reactive cung cấp một số công cụ để đồng bộ hóa các sự kiện không đồng bộ, cho phép các lập trình viên quản lý dễ dàng các sự kiện bất đồng bộ và giải quyết các vấn đề về đồng bộ hóa.

5. Xử lý dữ liệu lớn: System.Reactive cho phép các lập trình viên xử lý dữ liệu lớn một cách hiệu quả, cho phép lọc, chuyển đổi, kết hợp và thao tác với các luồng dữ liệu lớn.

6. Lập trình đa luồng: System.Reactive cung cấp các công cụ để quản lý các luồng đa nhiệm, giúp lập trình viên xây dựng các ứng dụng đa luồng một cách dễ dàng.

7. Điều khiển luồng dữ liệu: System.Reactive cho phép các lập trình viên điều khiển luồng dữ liệu, cho phép họ kiểm soát các sự kiện và thông báo được đưa vào hệ thống.

Trên đây chỉ là một số ứng dụng của System.Reactive trong C#, tùy thuộc vào nhu cầu sử dụng, các lập trình viên có thể áp dụng System.Reactive để giải quyết nhiều vấn đề khác nhau.

Một số ví dụ về sử dụng Reactive (Rx.NET):

Ví dụ 1: Tải dữ liệu từ http request api và hiển thị chúng vào Listbox

using Newtonsoft.Json;
using System;
using System.Collections.Generic;
using System.ComponentModel;
using System.Data;
using System.Drawing;
using System.Linq;
using System.Net.Http;
using System.Net.NetworkInformation;
using System.Reactive.Linq;
using System.Text;
using System.Threading.Tasks;
using System.Windows.Forms;

namespace DetectInternet
{
    public partial class Form1 : Form
    {
      
        public Form1()
        {
            InitializeComponent();
        }
        private readonly HttpClient httpClient = new HttpClient();
        private void Form1_Load(object sender, EventArgs e)
        {
            // Create an observable for a button click event
            var buttonClickObservable = Observable.FromEventPattern(button1, "Click");

            // Subscribe to the button click event
            buttonClickObservable.Subscribe(async _ =>
            {
                try
                {
                    // Send a GET request to the API
                    var response = await httpClient.GetAsync("https://jsonplaceholder.typicode.com/todos");

                    // Check if the response was successful
                    response.EnsureSuccessStatusCode();

                    // Read the response content as a string
                    var responseContent = await response.Content.ReadAsStringAsync();

                    // Deserialize the response content into an array of Todo objects
                    var todos = JsonConvert.DeserializeObject<Todo[]>(responseContent);

                    // Display the todos in a list box
                    listBox1.Items.Clear();
                    foreach (var todo in todos)
                    {
                        listBox1.Items.Add($"{todo.Id}: {todo.Title}");
                    }
                }
                catch (Exception ex)
                {
                    MessageBox.Show($"Failed to get todos: {ex.Message}");
                }
            });
        }

        private class Todo
        {
            public int Id { get; set; }
            public string Title { get; set; }
            public bool Completed { get; set; }
        }
    }
}

Khi chạy đoạn code trên kết quả, chúng ta sẽ được hình như ở bên dưới:

Ví dụ 2: Đây là một ví dụ về cách sử dụng System.Reactive để xử lý các sự kiện từ chuột và bàn phím trên một ứng dụng Windows Forms:

using System;
using System.Windows.Forms;
using System.Reactive.Linq;

namespace ReactiveWinForms
{
    public partial class MainForm : Form
    {
        public MainForm()
        {
            InitializeComponent();

            // Create observables for mouse and keyboard events
            var mouseMoveObservable = Observable.FromEventPattern<MouseEventArgs>(this, "MouseMove");
            var mouseClickObservable = Observable.FromEventPattern<MouseEventArgs>(this, "MouseClick");
            var keyDownObservable = Observable.FromEventPattern<KeyEventArgs>(this, "KeyDown");

            // Subscribe to mouse move events
            mouseMoveObservable
                .Throttle(TimeSpan.FromMilliseconds(100)) // Limit the frequency of mouse move events
                .Select(e => $"Mouse moved to ({e.EventArgs.X}, {e.EventArgs.Y})") // Convert to string message
                .Subscribe(msg => LogMessage(msg)); // Log message to the UI

            // Subscribe to mouse click events
            mouseClickObservable
                .Select(e => $"Mouse clicked at ({e.EventArgs.X}, {e.EventArgs.Y})") // Convert to string message
                .Subscribe(msg => LogMessage(msg)); // Log message to the UI

            // Subscribe to key down events
            keyDownObservable
                .Where(e => e.EventArgs.KeyCode == Keys.Escape) // Only handle escape key events
                .Select(e => "Escape key pressed") // Convert to string message
                .Subscribe(msg => LogMessage(msg)); // Log message to the UI
        }

        // Helper method to log a message to the UI
        private void LogMessage(string message)
        {
            if (logTextBox.InvokeRequired)
            {
                logTextBox.Invoke((MethodInvoker)(() => LogMessage(message)));
            }
            else
            {
                logTextBox.AppendText($"{message}\r\n");
            }
        }
    }
}

Trong ví dụ này, chúng ta sử dụng phương thức Observable.FromEventPattern() để tạo observables cho các sự kiện chuột và bàn phím.

Chúng ta cũng sử dụng phương thức Throttle() để giới hạn tần suất các sự kiện chuột di chuyển được phát ra.

Sau đó, chúng ta chuyển đổi các sự kiện thành chuỗi thông báo và đăng ký một subscriber để log các thông báo này lên giao diện người dùng.

Với cách tiếp cận này, chúng ta có thể dễ dàng xử lý các sự kiện từ chuột và bàn phím và áp dụng các phương thức xử lý theo ý muốn, ví dụ như giới hạn tần suất, lọc sự kiện theo điều kiện, hoặc chuyển đổi thành dạng thông báo để hiển thị cho người dùng.

Ví dụ 3:  Sử dụng System.Reactive để kiểm tra kết nối internet Realtime

rong ứng dụng Windows Form, chúng ta có thể sử dụng System.Reactive để phát hiện kết nối internet đã bị mất và cập nhật giao diện người dùng để thông báo cho người dùng biết.

Để làm được điều này, chúng ta có thể sử dụng lớp System.Net.NetworkInformation.NetworkChange và phương thức Observable.FromEvent() để tạo một observable từ sự kiện mất kết nối.

using System.Net.NetworkInformation;
using System.Reactive.Linq;

//...

var internetConnectionObservable = Observable.FromEvent<NetworkAvailabilityChangedEventHandler, bool>(
    h => (sender, e) => h(NetworkInterface.GetIsNetworkAvailable()),
    h => NetworkChange.NetworkAvailabilityChanged += h,
    h => NetworkChange.NetworkAvailabilityChanged -= h
);

internetConnectionObservable.Subscribe(isConnected =>
{
    if (isConnected)
    {
        // Internet connection has been restored
    }
    else
    {
        // Internet connection has been lost
    }
});

Ví dụ 4: Sử dụng System.Reactive để download file từ internet

Một ví dụ khác về cách sử dụng System.Reactive trong ứng dụng là tải xuống tệp tin từ internet.

Trong ví dụ này, chúng ta sử dụng System.Reactive để tạo một observable cho việc tải xuống tệp tin và theo dõi tiến trình tải xuống.

Để sử dụng System.Reactive trong ứng dụng này, ta sử dụng phương thức Observable.Create() để tạo một observable cho việc tải xuống tệp tin và sử dụng phương thức Subscribe() để đăng ký một subscriber để theo dõi tiến trình tải xuống.

using System.Reactive.Linq;
using System.Net;

//...

var client = new WebClient();
var downloadUrl = "http://example.com/file.zip";

var observableDownload = Observable.Create<byte[]>(async observer =>
{
    client.DownloadProgressChanged += (sender, e) =>
    {
        observer.OnNext(new byte[0]); // Progress event
    };

    client.DownloadDataCompleted += (sender, e) =>
    {
        observer.OnNext(e.Result); // Data completed event
        observer.OnCompleted();
    };

    var data = await client.DownloadDataTaskAsync(downloadUrl);
    observer.OnNext(data); // Data downloaded event
});

observableDownload.Subscribe(data =>
{
    // Handle downloaded data here
});

Trong đoạn mã trên, chúng ta sử dụng phương thức Observable.Create() để tạo một observable cho việc tải xuống tệp tin.

Chúng ta đăng ký các sự kiện DownloadProgressChanged, DownloadDataCompleted và sử dụng phương thức DownloadDataTaskAsync() để tải xuống dữ liệu từ URL được cung cấp.

Sau đó, chúng ta sử dụng phương thức Subscribe() để đăng ký một subscriber và xử lý dữ liệu được tải xuống.

Cách tiếp cận này cho phép chúng ta tải xuống các tệp tin từ internet một cách dễ dàng và đảm bảo rằng tiến trình tải xuống được theo dõi và các kết quả được xử lý đồng bộ.

Ví dụ 5: Sử dụng System.Reactive trong truy vấn SQL Server

Một ví dụ khác về cách sử dụng System.Reactive trong truy vấn SQL Server to DataGridView là sử dụng các phương thức Observable.Create(), BeginExecuteReader(), EndExecuteReader() và OnNext() để lấy dữ liệu và đẩy chúng vào một luồng dữ liệu:

using System;
using System.Data;
using System.Data.SqlClient;
using System.Reactive.Linq;
using System.Threading.Tasks;

public static class DatabaseHelper
{
    public static IObservable<DataTable> GetDataObservable(string connectionString, string sql)
    {
        return Observable.Create<DataTable>(async observer =>
        {
            using (var connection = new SqlConnection(connectionString))
            using (var command = new SqlCommand(sql, connection))
            {
                await connection.OpenAsync();

                var dataTable = new DataTable();
                var reader = await Task.Factory.FromAsync(
                    command.BeginExecuteReader, command.EndExecuteReader, null);

                dataTable.Load(reader);

                observer.OnNext(dataTable);
                observer.OnCompleted();
            }
        });
    }
}

Sau đó, ta có thể sử dụng phương thức SubscribeOn() để đăng ký một subscriber trên một luồng dữ liệu khác để tránh tắc nghẽn giao diện người dùng:

using System.Reactive.Concurrency;

//...

var subscription = DatabaseHelper.GetDataObservable(connectionString, sql)
    .SubscribeOn(TaskPoolScheduler.Default)
    .Subscribe(dataTable => dataGridView.DataSource = dataTable);

Trong đoạn mã trên, chúng ta sử dụng phương thức TaskPoolScheduler.Default để đăng ký một subscriber trên một luồng dữ liệu khác và sử dụng dataGridView.DataSource = dataTable để gán dữ liệu vào DataGridView.

Cách tiếp cận này cho phép chúng ta lấy dữ liệu từ SQL Server và hiển thị nó trên một DataGridView trong một ứng dụng Windows Forms một cách phản ứng và dễ dàng mở rộng cho các tác vụ khác trong ứng dụng của chúng ta.

Thanks for watching!

Tham khảo bới ChatGPT