.net Rx：消息的有序批处理

Question

我正在尝试使用 Rx 实现异步工作流，但我似乎完全错了。

我想做的是：

From an undefined asynchronous stream of un-parsed message strings (i.e. an IObservable<string>)
parse the message strings asynchronously, but preserve their order. (IObservable<Message>)
Batch up parsed Messages in groups of 100 or so (IObservable<IEnumerable<Message>>)
Send each batch, when complete, to the UI thread to be processed. Batches must arrive in the same order they were started.

我似乎无法获得订单保留，而且 Rx 似乎也没有像我期望的那样异步地做事。

我尝试使用 IEnumerable 而不是 IObservable 来保存订单，然后在其上调用 .AsParallel().AsOrdered() 运算符。这是代码。请参阅下面的注释了解我遇到的问题：

    private IObservable<IEnumerable<Message>> messageSource;
    public IObservable<IEnumerable<Message>> MessageSource { get { return messageSource; } }

    /// <summary>
    /// Sub-classes of MessageProviderBase provide this IEnumerable to 
    /// generate unparsed message strings synchronously
    /// </summary>
    protected abstract IEnumerable<string> UnparsedMessages { get; }

    public MessageProviderBase()
    {
        // individual parsed messages as a PLINQ query
        var parsedMessages = from unparsedMessage in UnparsedMessages.AsParallel().AsOrdered()
                                                 select ParseMessage(unparsedMessage);

        // convert the above PLINQ query to an observable, buffering up to 100 messages at a time
        var batchedMessages
            = parsedMessages.ToObservable().BufferWithTimeOrCount(TimeSpan.FromMilliseconds(200), 100);

        // ISSUE #1:
        // batchedMessages seems to call OnNext before all of the messages in its buffer are parsed.
        // If you convert the IObservable<Message> it generates to an enumerable, it blocks
        // when you try to enumerate it. 

        // Convert each batch to an IEnumerable
        // ISSUE #2: Even if the following Rx query were to run asynchronously (it doesn't now, see the above comment),
        // it could still deliver messages out of order. Only, instead of delivering individual
        // messages out of order, the message batches themselves could arrive out of order.
        messageSource = from messageBatch in batchedMessages
                                        select messageBatch.ToEnumerable().ToList();
    }

Answer 1

我下面的回答在某种程度上基于 Enigmativity 的代码，但修复了一些与完成相关的竞争条件，还增加了对取消和自定义调度程序的支持（这将使单元测试变得更加容易）。

public static IObservable<U> Fork<T, U>(this IObservable<T> source,
    Func<T, U> selector)
{
    return source.Fork<T, U>(selector, Scheduler.TaskPool);
}

public static IObservable<U> Fork<T, U>(this IObservable<T> source,
    Func<T, U> selector, IScheduler scheduler)
{
    return Observable.CreateWithDisposable<U>(observer =>
    {
        var runningTasks = new CompositeDisposable();

        var lockGate = new object();
        var queue = new Queue<ForkTask<U>>();
        var completing = false;
        var subscription = new MutableDisposable();

        Action<Exception> onError = ex =>
        {
            lock(lockGate)
            {
                queue.Clear();
                observer.OnError(ex);
            }
        };

        Action dequeue = () =>
        {
            lock (lockGate)
            {
                var error = false;
                while (queue.Count > 0 && queue.Peek().Completed)
                {
                    var task = queue.Dequeue();
                    observer.OnNext(task.Value);
                }
                if (completing && queue.Count == 0)
                {
                    observer.OnCompleted();
                }
            }
        };

        Action onCompleted = () =>
        {
            lock (lockGate)
            {
                completing = true;
                dequeue();
            }
        };

        Action<T> enqueue = t =>
        {
            var cancellation = new MutableDisposable();
            var task = new ForkTask<U>();

            lock(lockGate)
            {
                runningTasks.Add(cancellation);
                queue.Enqueue(task);
            }

            cancellation.Disposable = scheduler.Schedule(() =>
            {
                try
                {
                    task.Value = selector(t);

                    lock(lockGate)
                    {
                        task.Completed = true;
                        runningTasks.Remove(cancellation);
                        dequeue();
                    }
                }
                catch(Exception ex)
                {
                    onError(ex);
                }
            });
        };

        return new CompositeDisposable(runningTasks, 
            source.AsObservable().Subscribe(
                t => { enqueue(t); },
                x => { onError(x); },
                () => { onCompleted(); }
            ));
    });
}

private class ForkTask<T>
{
    public T Value = default(T);
    public bool Completed = false;
}

这是一个随机化任务执行时间来测试它的示例：

AutoResetEvent are = new AutoResetEvent(false);

Random rand = new Random();

Observable.Range(0, 5)
    .Fork(i =>
    {
        int delay = rand.Next(50, 500);
        Thread.Sleep(delay);

        return i + 1;
    })
    .Subscribe(
        i => Console.WriteLine(i),
        () => are.Set()
    );

are.WaitOne();

Console.ReadLine();

Answer 2

假设你有：

IObservable<string> UnparsedMessages = ...;
Func<string, Message> ParseMessage = ...;

然后你可以像这样使用SelectAsync 扩展方法：

IObservable<Message> ParsedMessages = UnparsedMessages.SelectAsync(ParseMessage);

SelectAsync 扩展方法异步处理每个未解析的消息，并确保结果按到达的顺序返回。

让我知道这是否满足您的需要。

代码如下：

public static IObservable<U> SelectAsync<T, U>(this IObservable<T> source,
    Func<T, U> selector)
{
    var subject = new Subject<U>();
    var queue = new Queue<System.Threading.Tasks.Task<U>>();
    var completing = false;
    var subscription = (IDisposable)null;

    Action<Exception> onError = ex =>
    {
        queue.Clear();
        subject.OnError(ex);
        subscription.Dispose();
    };

    Action dequeue = () =>
    {
        lock (queue)
        {
            var error = false;
            while (queue.Count > 0 && queue.Peek().IsCompleted)
            {
                var task = queue.Dequeue();
                if (task.Exception != null)
                {
                    error = true;
                    onError(task.Exception);
                    break;
                }
                else
                {
                    subject.OnNext(task.Result);
                }
            }
            if (!error && completing && queue.Count == 0)
            {
                subject.OnCompleted();
                subscription.Dispose();
            }
        }
    };

    Action<T> enqueue = t =>
    {
        if (!completing)
        {
            var task = new System.Threading.Tasks.Task<U>(() => selector(t));
            queue.Enqueue(task);
            task.ContinueWith(tu => dequeue());
            task.Start();
        }
    };

    subscription = source.Subscribe(
        t => { lock(queue) enqueue(t); },
        x => { lock(queue) onError(x); },
        () => { lock(queue) completing = true; });

    return subject.AsObservable();
}

---

我最终需要重新审视这个工作并编写了这个代码的更强大的版本（也基于 Richard 的回答。）

此代码的主要优点是没有任何显式队列。我纯粹是在使用任务延续来将结果按顺序排列。像款待一样工作！

public static IObservable<U> ForkSelect<T, U>(this IObservable<T> source, Func<T, U> selector)
{
    return source.ForkSelect<T, U>(t => Task<U>.Factory.StartNew(() => selector(t)));
}

public static IObservable<U> ForkSelect<T, U>(this IObservable<T> source, Func<T, Task<U>> selector)
{
    if (source == null) throw new ArgumentNullException("source");
    if (selector == null) throw new ArgumentNullException("selector");
    return Observable.CreateWithDisposable<U>(observer =>
    {
        var gate = new object();
        var onNextTask = Task.Factory.StartNew(() => { });
        var sourceCompleted = false;
        var taskErrored = false;

        Action<Exception> onError = ex =>
        {
            sourceCompleted = true;
            onNextTask = onNextTask.ContinueWith(t => observer.OnError(ex));
        };

        Action onCompleted = () =>
        {
            sourceCompleted = true;
            onNextTask = onNextTask.ContinueWith(t => observer.OnCompleted());
        };

        Action<T> onNext = t =>
        {
            var task = selector(t);
            onNextTask = Task.Factory.ContinueWhenAll(new[] { onNextTask, task }, ts =>
            {
                if (!taskErrored)
                {
                    if (task.IsFaulted)
                    {
                        taskErrored = true;
                        observer.OnError(task.Exception);
                    }
                    else
                    {
                        observer.OnNext(task.Result);
                    }
                }
            });
        };

        var subscription = source
            .AsObservable()
            .Subscribe(
                t => { if (!sourceCompleted) lock (gate) onNext(t); },
                ex => { if (!sourceCompleted) lock (gate) onError(ex); },
                () => { if (!sourceCompleted) lock (gate) onCompleted(); });

        var @return = new CompositeDisposable(subscription);

        return @return;
    });
}

并且允许使用 LINQ 的 SelectMany 重载是：

public static IObservable<U> SelectMany<T, U>(this IObservable<T> source, Func<T, Task<U>> selector)
{
    return source.ForkSelect<T, U>(selector);
}

public static IObservable<V> SelectMany<T, U, V>(this IObservable<T> source, Func<T, Task<U>> taskSelector, Func<T, U, V> resultSelector)
{
    if (source == null) throw new ArgumentNullException("source");
    if (taskSelector == null) throw new ArgumentNullException("taskSelector");
    if (resultSelector == null) throw new ArgumentNullException("resultSelector");
    return source.Zip(source.ForkSelect<T, U>(taskSelector), (t, u) => resultSelector(t, u));
}

所以现在可以像这样使用这些方法：

var observableOfU = observableOfT.ForkSelect(funcOfT2U);

或者：

var observableOfU = observableOfT.ForkSelect(funcOfT2TaskOfU);

或者：

var observableOfU =
    from t in observableOfT
    from u in funcOfT2TaskOfU(t)
    select u;

享受吧！