TPL Dataflow và Rx Ví dụ kết hợp

Question

Tôi chỉ muốn tìm hiểu cả hai và cách sử dụng chúng cùng nhau. Tôi hiểu rằng họ có thể bổ sung cho nhau, tôi không thể tìm thấy một ví dụ về một người thực sự làm điều đó.

Answer 1

Hãy để tôi bắt đầu với một chút nền.

.NET Framework có một số loại đặc biệt - một trong hai lớp học thích hợp hoặc giao diện - Task<T>, IObservable<T>, Nullable<T>, IEnumerable<T>, Lazy<T>, vv - cung cấp quyền hạn đặc biệt để loại cơ bản T.

TPL sử dụng Task<T> để thể hiện tính toán không đồng bộ của một giá trị đơn lẻ là T.

Rx sử dụng IObservable<T> để thể hiện tính toán không đồng bộ bằng không hoặc nhiều giá trị của T.

Đó là khía cạnh "tính toán không đồng bộ" của cả hai kết hợp này mang TPL và Rx cùng nhau.

Bây giờ, TPL cũng sử dụng loại Task để đại diện cho việc thực hiện không đồng bộ của một lambda Action, nhưng điều này có thể được coi là một trường hợp đặc biệt của Task<T> nơi T là void. Rất giống với một phương pháp tiêu chuẩn trong C# trả void như vậy:

public void MyMethod() { } 

Rx cũng cho phép đối với trường hợp đặc biệt cùng với việc sử dụng một loại đặc biệt gọi là Unit.

Sự khác biệt giữa TPL và Rx là số lượng giá trị được trả về. TPL là một và chỉ một trong khi Rx bằng 0 hoặc nhiều hơn. Vì vậy, nếu bạn đối xử với Rx một cách đặc biệt bằng cách chỉ làm việc với các chuỗi có thể quan sát trả về một giá trị duy nhất, bạn có thể thực hiện một số phép tính theo cách tương tự như TPL.

Ví dụ, trong TPL tôi có thể viết:

Task.Factory 
    .StartNew(() => "Hello") 
    .ContinueWith(t => Console.WriteLine(t.Result)); 

Và trong Rx tương đương sẽ là:

Observable 
    .Start(() => "Hello") 
    .Subscribe(x => Console.WriteLine(x)); 

tôi có thể đi một bước xa hơn trong Rx bằng cách xác định rằng TPL nên được sử dụng để thực hiện tính toán như sau:

Observable 
    .Start(() => "Hello", Scheduler.TaskPool) 
    .Subscribe(x => Console.WriteLine(x)); 

(Theo mặc định, Hồ bơi chủ đề được sử dụng.)

Bây giờ tôi có thể làm một số "trộn và kết hợp". Nếu tôi thêm một tham chiếu đến không gian tên System.Reactive.Threading.Tasks tôi có thể di chuyển giữa các tác vụ và các quan sát khá dễ dàng.

Task.Factory 
    .StartNew(() => "Hello") 
    .ToObservable() 
    .Subscribe(x => Console.WriteLine(x)); 

Observable 
    .Start(() => "Hello") 
    .ToTask() 
    .ContinueWith(t => Console.WriteLine(t.Result)); 

Thông báo các ToObservable() & .ToTask() cuộc gọi và kết quả flips từ một thư viện để người kia.

Nếu tôi có thể quan sát trả về nhiều giá trị, tôi có thể sử dụng phương pháp mở rộng có thể quan sát để biến nhiều giá trị chuỗi thành một giá trị mảng có thể được chuyển thành công việc.Giống như vậy:

Observable 
    .Interval(TimeSpan.FromSeconds(1.0)) 
    .Take(5) // is IObservable<long> 
    .ToArray() 
    .ToTask() // is Task<long[]> 
    .ContinueWith(t => Console.WriteLine(t.Result.Length)); 

Tôi nghĩ đây là câu trả lời khá cơ bản cho câu hỏi của bạn. Đó là những gì bạn đang mong đợi?