使用Rx库探索异步编程：.NET Reactive Extensions 实战

需积分: 9 44 浏览量更新于2024-07-26 收藏 2.09MB PDF 举报

"Rx HOL .NET 是一个针对.NET平台的Reactive Extensions（Rx）的实践教程，通过一系列逐步深入的示例，帮助读者理解和掌握如何使用Rx来编写异步应用程序，利用可观察集合的概念解决异步编程的复杂性。本教程适合对.NET和C#有基础、熟悉异步编程概念的开发者，需要Visual Studio 2010及.NET 4的环境，并需安装Rx for .NET 4的扩展库。" Rx（Reactive Extensions）是微软开发的一个库，它引入了可观察集合的概念，用于构建异步和基于事件的程序。这个框架的核心特性体现在以下三个方面： 1. 可观察集合（Observable Collections）：Rx将数据流和事件视为可观察的对象，就像普通的集合一样可以进行操作，如过滤、映射、合并等。这使得处理异步数据流变得更加直观和灵活。 2. 异步编程（Asynchronous Programming）：Rx提供了一种统一的方式来处理异步操作，无论是基于回调、基于任务（Task）还是基于事件。它通过IObservable和IObserver接口抽象出异步操作的生产者和消费者，简化了异步代码的编写和管理。 3. 组合性（Composability）：Rx的强大之处在于其组合性。开发者可以将多个可观察序列组合成更复杂的逻辑，这种模块化的方法可以降低代码的复杂度，提高可读性和可维护性。在实践中，Rx HOL .NET将通过一系列的示例指导读者： - 学习创建和订阅可观察序列的基本用法。 - 掌握如何使用 LINQ 查询操作符处理可观察序列，如Select、Where、Concat、Merge等。 - 了解延迟和缓存机制，如Debounce和Throttle，用于处理连续快速的事件。 - 探索错误处理和异常传播，以及如何优雅地终止序列。 - 学习如何使用Scheduler来控制操作的调度，以便在不同的线程或时间点执行。 - 理解如何使用Zip、CombineLatest等操作符同步多个数据源。通过完成这些示例，开发者将能够更好地理解Rx如何通过可观察集合的概念来简化异步编程，提高代码的组织性和可测试性，同时也为处理复杂的并发和实时数据流问题提供了强大工具。对于那些在.NET环境中进行异步编程的开发者来说，掌握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页未读，继续阅读

atskyline

粉丝: 27
资源: 8

使用Rx库探索异步编程：.NET Reactive Extensions 实战

响应式编程Rx.net英文入门电子书Rx HOL .NET C#

Rx+HOL+.NET.pdf

HTTP/2协议存在什么问题？

matlab实现gs算法生成全息图

CryptoVerify 资料

CryptHOL语言是什么？举个例子

厦门理工学院在广东2021-2024各专业最低录取分数及位次表.pdf

算法竞赛学习资源114514

基于ssm汽车在线销售系统设计与实现.docx

河南中医药大学在广东2021-2024各专业最低录取分数及位次表.pdf

最新资源