0

目的是在 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();
}

问题 这是背压吗？如果不是，是否还有其他相关术语？

4

1 回答 1

2

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

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

于 2018-03-22T13:32:32.757 回答