RX.NET-过程组异步并与受约束的并发并行

Question

播放系统。试图解决下一个任务的反应性 -

• 将传入的字符串流分成
• 每个组中的小组项目必须异步处理和顺序处理
• 组，必须并行处理
• 不超过n组，同时必须同时处理N组
• 。到目前为止，使用Sync Pirinitives使用同步原始

图是我到目前为止发现的最好的 -

TaskFactory taskFactory = new (new LimitedConcurrencyLevelTaskScheduler(2));
TaskPoolScheduler scheduler = new (taskFactory);
source
    .GroupBy(item => item)
    .SelectMany(g => g.Select(item => Observable.FromAsync(() => onNextAsync(item))).ObserveOn(scheduler).Concat())
    .Subscribe();

任何想法如何使用OA调度程序来实现它？无法通过Merge（）使其正常工作

Answer 1

执行的最简单方法“不得同时处理n组” 限制，可能是使用 Smaphoreslim 。因此，

.SelectMany(g => g.Select(item => Observable.FromAsync(() => onNextAsync(item))).Concat())

您可以做到这一点：您可以做到：

var semaphore = new SemaphoreSlim(N, N);

//...

.SelectMany(g => g.Select(item => Observable.FromAsync(async () =>
{
    await semaphore.WaitAsync();
    try { return await onNextAsync(item); }
    finally { semaphore.Release(); }
})).Merge(1))

btw in 当前Rx版本（5.0.0） concat 运算符，我更喜欢使用 Merge（1）而不是。

Answer 2

为了使用Rx工具解决此问题，理想情况下，您希望拥有类似的内容：

source
    .GroupBy(item => item.Key)
    .Select(group => group.Select(
        item => Observable.FromAsync(() => ProcessAsync(item))).Merge(1))
    .Merge(maxConcurrent: N)
    .Wait();

Inner Merge（1）将在每个组中强制执行独家处理，而外部 Merge（n） ）将执行全球最大并发策略。不幸的是，这是不起作用的，因为外部合并（n）将订阅限制为内部序列（ igroupedObservable＆lt; t＆gt; s），而不是其各个元素。这不是您想要的。结果是只有要处理的第一个N组，并且所有其他组的元素都将被忽略。 groupby 操作员会创建热门子序列，如果您不立即订阅它们，您将丢失元素。

为了根据需要使用外部合并（n），您必须自由合并由 observable.fromasync 的所有内部序列合并，并且具有其他一些机制来序列化每个组的处理。一个想法是实现一个特殊的选择运算符，该操作员仅在上一个完成后才 observable.fromasync 。以下是基于 zip 操作员的实现。 zip 操作员在内部维护两个隐藏的缓冲区，因此它可以从两个序列中产生对以不同频率散发元素的对。为了避免丢失元素，我们正是我们需要的。

private static IObservable<IObservable<TResult>> SelectOneByOne<TSource, TResult>(
    this IObservable<TSource> source,
    Func<TSource, IObservable<TResult>> selector)
{
    var subject = new BehaviorSubject<Unit>(default);
    var synchronizedSubject = Observer.Synchronize(subject);
    return source
        .Zip(subject, (item, _) => item)
        .Select(item => selector(item).Do(
            _ => { },
            _ => synchronizedSubject.OnNext(default),
            () => synchronizedSubject.OnNext(default)));
}

civationUbject＆lt; t＆gt; 最初包含一个元素，因此第一对将立即产生。在处理第一个元素之前，第二对不会产生。与第三对和第二个元素等相同。

然后，​​您可以使用此操作员来解决这样的问题：

source
    .GroupBy(item => item.Key)
    .SelectMany(group => group.SelectOneByOne(
        item => Observable.FromAsync(() => ProcessAsync(item))))
    .Merge(maxConcurrent: N)
    .Wait();

上述解决方案仅出于回答问题而呈现。我认为我不会在生产环境中信任它。