RX terminolgy:当存在频繁的可观察通知时，RX 运算符中的异步处理

Question

目的是在 RX 运算符(例如 Select)中对稀缺资源执行一些异步工作。当可观察通知的速率快于异步操作完成所需的时间时，会出现问题。

现在我实际上解决了这个问题。我的问题是，对于这种特定类型的问题，正确的术语是什么？它有名字吗？是背压吗？到目前为止，我所做的研究表明，这是一个压力问题，但根据我的理解，不一定是背压。我找到的最相关的资源是这些： https://github.com/ReactiveX/RxJava/wiki/Backpressure-(2.0) http://reactivex.io/documentation/operators/backpressure.html

现在进入实际代码。假设有一种稀缺资源，它是消费者。在这种情况下，当资源正在使用时引发异常。请注意，此代码不应更改。

public class ScarceResource
{
private static bool inUse = false;
public async Task<int> AccessResource()
{
if (inUse) throw new Exception("Resource is alredy in use");
var result = await Task.Run(() =>
{
inUse = true;
Random random = new Random();
Thread.Sleep(random.Next(1, 2) * 1000);
inUse = false;
return random.Next(1, 10);
});
return result;
}
}
public class ResourceConsumer
{
public IObservable<int> DoWork()
{
var resource = new ScarceResource();
return resource.AccessResource().ToObservable();
}
}

现在，这是使用资源的幼稚实现的问题。引发错误是因为通知的发送速度快于使用者运行的速度。

private static void RunIntoIssue()
{
var numbers = Enumerable.Range(1, 10);
var observableSequence = numbers
.ToObservable()
.SelectMany(n =>
{
Console.WriteLine("In observable: {0}", n);
var resourceConsumer = new ResourceConsumer();
return resourceConsumer.DoWork();
});
observableSequence.Subscribe(n => Console.WriteLine("In observer: {0}", n));
}

使用以下代码解决了问题。我通过将已完成的 BehaviorSubject 与 Zip 运算符结合使用来减慢处理速度。从本质上讲，此代码的作用是采用顺序方法而不是并行方法。

private static void RunWithZip()
{
var completed = new BehaviorSubject<bool>(true);
var numbers = Enumerable.Range(1, 10);
var observableSequence = numbers
.ToObservable()
.Zip(completed, (n, c) =>
{
Console.WriteLine("In observable: {0}, completed: {1}", n, c);
var resourceConsumer = new ResourceConsumer();
return resourceConsumer.DoWork();
})
.Switch()
.Select(n =>
{
completed.OnNext(true);
return n;
});
observableSequence.Subscribe(n => Console.WriteLine("In observer: {0}", n));
Console.Read();
}

问题这是背压吗，如果不是，它是否有另一个相关的术语？

Answer 1

你基本上是在实现一种形式的锁定或互斥锁。你的代码会导致背压，它并没有真正处理它。

想象一下，如果你的源不是生成器函数，而是一系列数据推送。数据推送以每毫秒的恒定速率到达。处理每个需要 10 个 Millis，并且您的代码强制串行处理。这会导致背压：Zip 将无限期地将未处理的数据推送排队，直到内存不足。