System.Reactive：如何缓冲Observable即时值与后续单值？

这个需求完全可以实现，核心思路是利用Rx中同步推送和异步推送的执行时机差异：当你订阅一个Observable时，所有同步产生的OnNext会在Subscribe方法调用期间立即执行，而异步推送则会在后续的调度器上触发。我们可以通过这个特性，把初始的所有即时值打包成第一个数组，之后每个异步推送的值单独包装成数组。

实现自定义扩展方法 #

我们可以写一个通用的扩展方法BufferInitialBatch，适用于任意IObservable<T>：

public static class ObservableExtensions
{
    public static IObservable<T[]> BufferInitialBatch<T>(this IObservable<T> source)
    {
        return Observable.Create<T[]>(observer =>
        {
            var initialBatch = new List<T>();
            bool hasEmittedInitialBatch = false;

            // 订阅源Observable，先收集所有同步推送的初始值
            var subscription = source.Subscribe(
                item =>
                {
                    if (!hasEmittedInitialBatch)
                    {
                        // 同步阶段：收集初始值
                        initialBatch.Add(item);
                    }
                    else
                    {
                        // 异步阶段：推送单个元素的数组
                        observer.OnNext(new[] { item });
                    }
                },
                error =>
                {
                    // 出错时，先推送已收集的初始值（如果有的话），再转发错误
                    if (!hasEmittedInitialBatch)
                    {
                        if (initialBatch.Any())
                        {
                            observer.OnNext(initialBatch.ToArray());
                        }
                        hasEmittedInitialBatch = true;
                    }
                    observer.OnError(error);
                },
                () =>
                {
                    // 完成时，推送剩余的初始值（如果有的话），再转发完成信号
                    if (!hasEmittedInitialBatch)
                    {
                        if (initialBatch.Any())
                        {
                            observer.OnNext(initialBatch.ToArray());
                        }
                        hasEmittedInitialBatch = true;
                    }
                    observer.OnCompleted();
                }
            );

            // Subscribe方法执行完毕，所有同步推送的初始值已收集完成，推送初始批次
            if (!hasEmittedInitialBatch)
            {
                if (initialBatch.Any())
                {
                    observer.OnNext(initialBatch.ToArray());
                }
                hasEmittedInitialBatch = true;
            }

            return subscription;
        });
    }
}

验证你的示例场景 #

用你给出的测试代码验证：

var immediate_values = new [] { "currently", "available", "values" }.ToObservable();
var future_values = Observable.Timer(TimeSpan.FromSeconds(5), TimeSpan.FromSeconds(1))
                              .Select(x => "new value!");
IObservable<string> input = immediate_values.Concat(future_values);

IObservable<string[]> buffered = input.BufferInitialBatch();

buffered.Subscribe(arr => 
{
    Console.WriteLine($"Received: [{string.Join(", ", arr)}]");
});

预期输出：

Received: [currently, available, values]
// 等待5秒后
Received: [new value!]
// 每秒一次
Received: [new value!]
...

边界情况说明 #

1. 无初始即时值：如果源Observable没有同步推送的值，初始批次不会推送空数组，而是直接等待第一个异步值并推送单个元素数组。
2. 初始值为空：如果源Observable同步推送了0个值，同样不会推送空数组，直接进入异步值处理阶段。
3. 中途出错：如果在同步收集初始值时发生错误，会先推送已收集到的初始值，再转发错误信号；如果在异步阶段出错，直接转发错误。

为什么不用内置的Buffer方法？ #

内置的Buffer系列方法需要明确的窗口关闭信号（比如时间、计数、另一个Observable），但你的需求无法提前确定初始值的数量或结束时机——而利用Rx的同步/异步执行特性，我们可以自动捕获所有即时值，不需要额外的信号源。

内容的提问来源于stack exchange，提问作者Bogey