使用Reactive X进行异步编程：MSDN实战教程

需积分: 10 82 浏览量更新于2024-07-26 1 收藏 2.02MB PDF 举报

"这篇资源是关于Reactive X的MSDN教程，旨在帮助读者逐步学习和理解Reactive Extensions for .NET (Rx)。通过一系列渐进的示例，读者可以掌握Rx如何利用可观察集合的概念来编写异步应用程序，并体验其组合能力的威力。" 在深入探讨Reactive X之前，我们需要了解一些先决条件。首先，要求读者具备.NET和C#编程语言的活跃知识，因为Rx主要与这些技术相结合。其次，对异步编程的概念及其复杂性有深刻的理解也是必要的，这将有助于更好地吸收Rx的核心理念。此外，实验环境需要Visual Studio 2010和.NET Framework 4（尽管可以使用早期版本，但实验室内容是针对VS2010设计的），并确保已经安装了Rx for .NET 4，可以从MSDN DevLabs获取。那么，什么是Reactive X呢？Rx可以用一句话概括：它是一个使用可观察集合来构建异步和基于事件的程序的库。这个定义揭示了Rx的三个核心属性： 1. 可观察集合：Rx引入了一个新的抽象概念——可观察序列，它类似于传统的集合，但能发出值，而不是简单地包含值。这些序列可以被订阅，当它们变化时，订阅者会收到通知。这种模式使得处理事件流和数据流变得非常直观。 2. 异步编程：Rx允许开发者以声明式的方式处理异步操作，而不是传统的回调或基于状态的编程。这种方式降低了异步代码的复杂性，提高了可读性和可维护性。 3. 组合能力：Rx的强大之处在于它的组合能力。通过操作符，可以组合多个可观察序列，过滤、映射、合并或转换它们，创建复杂的异步逻辑，而无需嵌套回调或手动管理线程。在本教程中，读者将逐步接触以下关键概念和操作符： - 创建可观察序列：例如，使用`Observable.Create`或`Observable.FromAsyncPattern`将异步操作转化为可观察序列。 - 数据变换：如`Select`用于映射每个元素，`Where`用于过滤序列中的元素。 - 错误处理：通过`Catch`和`Retry`操作符处理异常和重试逻辑。 - 合并与组合：`Merge`和`Zip`操作符可以将多个序列合并成一个，`Concat`则按顺序合并。 - 调度器：使用调度器控制序列的订阅和通知在哪个线程上执行，这对于UI线程同步至关重要。通过这个实验室，读者将不仅学习到Rx的基本用法，还会体验到它如何简化异步编程的挑战，以及如何通过可观察序列的组合来构建强大的、响应式的应用。随着对Rx的理解加深，开发者能够更高效地处理事件驱动和异步编程场景，提升软件的性能和用户体验。

5. One final essential factory method or primitive constructor is called Return. Its role is to represent a single-

element sequence, just like a single-cell array would be in the world of IEnumerable sequences. The behavior

observed by subscribed observers is two messages: OnNext with the value and OnCompleted signaling the end

of the sequence has been received:

IObservable<int> source = Observable.Return(42);

Running this code produces the following output:

OnNext: 42

OnCompleted

Background: Return plays an essential role in the theory behind LINQ, known as monads. Together with an

operator called SelectMany (which we’ll learn about more later on), they form the primitive functions needed to

leverage the power of monadic computation. More information can be found by searching the web using monad

and functional as the keywords.

6. At this point, we’ve seen the most trivial observable sequence constructors that are intimately related to an

observer’s triplet of methods. While those are of interest in certain cases, more meaty sequences are worth to

explore as well. The Range operator is just one operator that generates such a sequence. Symmetric to the same

operator on Enumerable, Range does return a sequence with 32-bit integer values given a starting value and a

length:

IObservable<int> source = Observable.Range(5, 3);

Running this code produces the following output:

OnNext: 5

OnNext: 6

OnNext: 7

OnCompleted

Note: As with all the sequences mentioned in this exercise, Range is a cold observable. To recap, this simply

means that it starts producing its results to an observer upon subscription. Another property of cold observable

sequence is that every subscription will cause such reevaluation. Thus, if two calls to Subscribe are made, both

of the observers passed to those calls will receive the messages from the observable. It’s not because the data

observation has run to completion for one observer that other observers won’t run anymore. Whether or not

the produced data is the same for every observer depends on the characteristics of the sequence that’s being

generated. For deterministic and “purist functional” operators like Return and Range, the messages delivered to

every observer will be the same. However, one could imagine other kinds of observable sequences that depend

on side-effects and thus deliver different results for every observer that subscribes to them.

7. To generalize the notion of sequence creation in terms of a generator function, the Generate constructor

function was added to Rx. It closely resembles the structure of a for-loop as one would use to generate an

enumerable sequence using C# iterators (cf. the “yield return” and “yield break” statements). To do so, it takes a

number of delegate-typed parameters that expect function to check for termination, to iterate one step and to

emit a result that becomes part of the sequence and is sent to the observer:

IObservable<int> source = Observable.Generate(0, i => i < 5, i => i + 1, i => i * i);

Running this code produces the following output:

OnNext: 0

OnNext: 1

OnNext: 4

OnNext: 9

OnNext: 16

OnCompleted

Note: A sister function called GenerateWithTime exists that waits a computed amount of time before moving

on to the next iteration. We’ll look at this one in just a moment.

8. One operator that may seem interesting from a curiosity point of view only is called Never. It creates an

observable sequence that will never signal any notification to a subscribed observer:

IObservable<int> source = Observable.Never<int>();

Running this code shouldn’t produce any output till the end of mankind.

Background: One reason this operator has a role is to reason about composition with other sequences, e.g.

what would it mean to concatenate a finite and an infinite sequence? Should the result also exhibit an infinite

characteristic? Those answers are essential to ensuring all of the Rx semantics make sense and are consistent

throughout various operators. Besides the operator’s theoretical use, there are real use cases for it as well. For

example, you may use it to ensure a sequence never terminates by avoiding an OnCompleted handler getting

triggered. Another use is to test whether an application does not hang in the presence of non-termination.

9. Where the Subscribe method has an asynchronous characteristic, other blocking operators exist to observe a

sequence in a synchronous manner. One such operator is called Run.

First of all, let’s formalize “asynchronous”. It’s near to impossible to say something is asynchronous in itself

without mentioning two parties. In particular, the thread calling the Subscribe method on an observable

sequence is not blocked till the sequence runs to completion, which may happen on another thread. That is,

Subscribe is asynchronous in that the caller is not blocked till the observation of the sequence completes.

For demos and testing it’s often useful to perform a “blocking subscribe” such that the caller is blocked till the

observable sequence triggers OnError or OnCompleted to the observer. That’s what Run is all about:

IObservable<int> source = Observable.Range(0, 10);

source.Run(

x => Console.WriteLine("OnNext: {0}", x),

ex => Console.WriteLine("OnError: {0}", ex.Message),

() => Console.WriteLine("OnCompleted")

);

In here, we won’t get to the next statement till the sequence completes gracefully or exceptionally.

剩余41页未读，继续阅读

PerfectBoyHHH

粉丝: 0
资源: 6

使用Reactive X进行异步编程：MSDN实战教程

RxJava 2.x 教程（完结版）1

Reactive-Programming-in-Java-Using-RxJava-3.x-ReactiveX:使用RxJava 3.x在Java中进行React式编程

spring-webflux-reactive-rest-api-demo：使用Spring WebFlux和Reactive Mongo构建Reactive Rest API

ref reactive使用区别

vue.reactive方法

vue3 reactive的坑

vue3,reactive与Proxy

vue3reactive函数

vue中的Ref 和 Reactive

value cannot be made reactive

最新资源