Reactive.Subject 让 onNext 等待上一个 onNext 动作
Reactive.Subject make onNext wait for previous onNext action
我正在尝试实现可观察的等待 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
我正在尝试实现可观察的等待 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