如何關聯功能輸入和輸出？

Question

考慮下面的簡單程序。它具有可觀測的整數和函數來計算最近發佈的整數是偶數還是奇數。出乎意料的是，該程序報告最近的數字是否是偶數/之前它報告數字已更改。

static void Main(string[] args) { 
    int version = 0; 
    var numbers = new Subject<int>(); 
    IObservable<bool> isNumberEven = numbers.Select(i => i % 2 == 0); 
    isNumberEven 
     .Select(i => new { IsEven = i, Version = Interlocked.Increment(ref version) }) 
     .Subscribe(i => Console.WriteLine($"Time {i.Version} : {i.IsEven}")); 
    numbers 
     .Select(i => new { Number = i, Version = Interlocked.Increment(ref version) }) 
     .Subscribe(i => Console.WriteLine($"Time {i.Version} : {i.Number}")); 
    numbers.OnNext(1); 
    numbers.OnNext(2); 
    numbers.OnNext(3); 
    Console.ReadLine(); 
} 

的輸出是：

Time 1 : False 
Time 2 : 1 
Time 3 : True 
Time 4 : 2 
Time 5 : False 
Time 6 : 3 

我認爲改變數目將掀起的下游效應的級聯，並且這些將在訂單它們發生被報告。交換訂閱訂單將交換報告結果的方式。我知道rx是異步的，事情可能以非確定性的順序發生。如果我在函數中使用.Delay（）或Web調用，我不能確定何時會報告結果。但在這種情況下，我感到非常驚訝。

爲什麼這是一件大事？我認爲這意味着如果我想嘗試關聯函數輸入和輸出（如發佈時的打印數字以及它們是偶數還是奇數），則必須在輸出結果中包含輸入參數，如下所示：

var isNumberEven = numbers.Select(i => new { 
    Number = i, 
    IsEven = i % 2 == 0 
}); 

我想我可以構建一堆小的簡單函數，然後使用rx操作符來編寫它們來完成複雜的計算。但也許我不能使用rx運算符來組合/連接/關聯結果。當我定義每個功能時，我必須關聯輸入和輸出。

在某些情況下，我可以使用rx運算符來關聯結果。如果每個輸入都生成一個輸出，我可以壓縮這兩個輸入。但是一旦你做了像調節輸入那樣的東西，它就不再工作了。

該版本的程序似乎報告數字是否以合理的方式是偶數或奇數。

static void Main(string[] args) { 
    var numbers = new Subject<int>(); 
    var isNumberEven = numbers.Select(i => i % 2 == 0); 
    var publishedNumbers = numbers.Publish().RefCount(); 
    var report = 
     publishedNumbers 
     .GroupJoin(
      isNumberEven, 
      (_) => publishedNumbers, 
      (_) => Observable.Empty<bool>(), 
      (n, e) => new { Number = n, IsEven = e }) 
     .SelectMany(i => i.IsEven.Select(j => new { Number = i.Number, IsEven = j })); 
    report.Subscribe(i => Console.WriteLine($"{i.Number} {(i.IsEven ? "even" : "odd")}")); 
    numbers.OnNext(1); 
    numbers.OnNext(2); 
    numbers.OnNext(3); 
    Console.ReadLine(); 
} 

輸出看起來像：

1 odd 
2 even 
3 odd 

但我不知道這是否是一個幸運的巧合還是我可以依靠它。 Rx中的哪些操作按確定性順序發生？哪些是不可預測的？我應該定義所有的函數來將輸入參數包含在結果中嗎？

Answer 1

您的第一個程序的行爲與我所期望的完全相同，並且確定如此。

我明白rx是異步的，它可能會以非確定性的順序發生。

如果引入非確定性行爲（如併發性/調度），則事件只會以非確定性順序發生，否則Rx是確定性的。

在這裏玩有幾個問題/誤解。 1）可變的外部狀態 - version 2）使用主題（但在這個示例中根本不是問題） 3）誤解了回調的發佈方式。

讓我們只關注3）。如果我們把你的代碼和解包到它的基本調用，你可能會看到Rx是多麼的簡單。

numbers.OnNext(1);該主題將按照訂閱的順序查找訂閱和OnNext。

IObservable<bool> isNumberEven = numbers.Select(i => i % 2 == 0); 
isNumberEven 
    .Select(i => new { IsEven = i, Version = Interlocked.Increment(ref version) }) 
    .Subscribe(i => Console.WriteLine($"Time {i.Version} : {i.IsEven}")); 

也可以減少到

numbers.Select(i => i % 2 == 0) 
    .Select(i => new { IsEven = i, Version = Interlocked.Increment(ref version) }) 
    .Subscribe(i => Console.WriteLine($"Time {i.Version} : {i.IsEven}")); 

一個可以認爲，隨着isNumberEven從未使用過其他地方，你應該它減少到這一點。

所以我們可以看到我們有我們的第一個用戶。 並有效地將運行的代碼是這樣的

private void HandleOnNext(int i) 
{ 
    var isEven = i % 2 == 0 
    var temp = new { IsEven = isEven , Version = Interlocked.Increment(ref version) }; 
    Console.WriteLine($"Time {temp .Version} : {temp .IsEven}"); 
} 

我們的第二個用戶（因爲.Subscribe(方法偶數訂閱後的稱呼），是numbers用戶。 他的代碼可以有效地歸結爲

private void HandleOnNext(int i) 
{ 
    var temp = new { Number = i, Version = Interlocked.Increment(ref version) }; 
    Console.WriteLine($"Time {temp.Version} : {temp.Number}"); 
} 

所以一旦你已經完全解構的代碼，你最終基本上這

void Main() 
{ 
    int version = 0; 

    //numbers.OnNext(1); 
    ProcessEven(1, ref version); 
    ProcessNumber(1, ref version); 
    //numbers.OnNext(2); 
    ProcessEven(2, ref version); 
    ProcessNumber(2, ref version); 
    //numbers.OnNext(3); 
    ProcessEven(3, ref version); 
    ProcessNumber(3, ref version); 
} 

// Define other methods and classes here 
private void ProcessEven(int i, ref int version) 
{ 
    var isEven = i % 2 == 0; 
    var temp = new { IsEven = isEven, Version = Interlocked.Increment(ref version) }; 
    Console.WriteLine($"Time {temp.Version} : {temp.IsEven}"); 
} 
private void ProcessNumber(int i, ref int version) 
{ 
    var temp = new { Number = i, Version = Interlocked.Increment(ref version) }; 
    Console.WriteLine($"Time {temp.Version} : {temp.Number}"); 
} 

一旦所有的回調和訂閱物化，那麼你就可以看到這不是神奇的事情，一切都是確定性的。

我應該定義所有的函數來將輸入參數包含在結果中嗎？

要回答你的問題（因爲你對Rx的誤解我很猶豫，你只需要這樣做，當結果序列的順序是非確定性的。 這樣做的一個例子可能是您一次發出多個Web請求。 您不能確定他們會按照您發送的順序進行回覆。 但是，您可以強制這些場景回到與運營商的使用，如Concat