Reactive.Subject 让 onNext 等待上一个 onNext 动作

问题描述

我试图实现可观察的等待 onNext 操作完成，然后再继续下一个操作。我发现工作的唯一方法是使用 SemaphoreSlim。 Reactive 有没有办法在不使用 SemaphoreSlim 的情况下做到这一点？我找不到。

 SemaphoreSlim _semaphoreSlim = new SemaphoreSlim(1,1);
  subject.Window(() => subject.Throttle(TimeSpan.FromMilliseconds(500)))
            .SelectMany(c => c.ToList())
            .Subscribe(async x =>
                {
                    await _semaphoreSlim.WaitAsync();
                    try
                    {
                        //await Async code here
                    }
                    finally
                    {
                        _semaphoreSlim.Release();
                    }
                });

解决方法

首先，我创建了一个示例 Media_Load_Async 来模拟您正在做的事情。

static int max = 5;
static Random random = new Random();
static TimeSpan[] delays = Enumerable.Range(0,max).Select(x => TimeSpan.FromSeconds(random.Next(5) + 2)).ToArray();

async Task<Unit> Media_Load_Async(int index,TimeSpan delay)
{
    Console.WriteLine($"Start {index} - {delay}");
    await Task.Delay(delay);
    Console.WriteLine($"End {index}");
    return Unit.Default;
}

如果我运行这个简单的查询，那么我会得到你试图避免的结果：

Observable
    .Range(0,max)
    .Subscribe(async x => await Media_Load_Async(x,delays[x]));

Start 0 - 00:00:05
Start 1 - 00:00:06
Start 2 - 00:00:02
Start 3 - 00:00:06
Start 4 - 00:00:04
End 2
End 4
End 0
End 3
End 1

如果我输入您的 SemaporeSlim 代码，我会得到您想要的：

SemaphoreSlim _semaphoreSlim = new SemaphoreSlim(1,1);
Observable
    .Range(0,max)
    .Subscribe(async x =>
    {
        await _semaphoreSlim.WaitAsync();
        try
        {
            await Media_Load_Async(x,delays[x]);
        }
        finally
        {
            _semaphoreSlim.Release();
        }
    });

Start 0 - 00:00:04
End 0
Start 1 - 00:00:03
End 1
Start 2 - 00:00:06
End 2
Start 3 - 00:00:04
End 3
Start 4 - 00:00:03
End 4

接下来，如果我将对 Media_Load_Async 的调用移动到查询本身内部，那么我又回到了原来的问题：

Observable
    .Range(0,max)
    .SelectMany(x => Observable.FromAsync(() => Media_Load_Async(x,delays[x])))
    .Subscribe();

Start 0 - 00:00:06
Start 1 - 00:00:02
Start 2 - 00:00:06
Start 3 - 00:00:05
Start 4 - 00:00:02
End 4
End 1
End 3
End 0
End 2

但是，如果我将 SelectMany 换成 Select/Concat 对，那么无需 SemaphoreSlim 即可得到您想要的：

Observable
    .Range(0,max)
    .Select(x => Observable.FromAsync(() => Media_Load_Async(x,delays[x])))
    .Concat()
    .Subscribe();

Start 0 - 00:00:04
End 0
Start 1 - 00:00:05
End 1
Start 2 - 00:00:02
End 2
Start 3 - 00:00:06
End 3
Start 4 - 00:00:06
End 4