如何基于缓冲项的条件使用Observable.Buffer进行序列缓冲？

我想根据依赖当前项的条件对序列中的元素进行缓冲，规则如下：

• 遇到奇数时停止缓冲
• 遇到偶数时开始缓冲

给定输入序列：

new[] { 1, 3, 5, 7, 2, 4, 6, 8, 1 };

期望输出的缓冲结果：

{ 1 }
{ 3 }
{ 5 }
{ 7 }
{ 2, 4, 6, 8 }
{ 1 }

我尝试了以下代码但没有得到预期结果：

var boundaries = origin.Select(x => x % 2 != 0).DistinctUntilChanged();
var result = origin.Buffer(boundaries);

你原代码的问题在于，Buffer(boundaries)的边界逻辑不符合你的分组规则：Buffer会把边界信号作为分隔点，但你的需求是奇数单独成组、偶数连续成组，原逻辑没处理好分组的起始与结束时机。

方法一：手动遍历分组（直观易懂） #

如果不需要依赖Rx的响应式操作符，直接遍历序列手动维护缓冲是最清晰的方式：

var origin = new[] { 1, 3, 5, 7, 2, 4, 6, 8, 1 };
var result = new List<List<int>>();
List<int> currentBuffer = null;

foreach (var num in origin)
{
    if (num % 2 != 0)
    {
        // 遇到奇数：若有未完成的偶数缓冲，先提交；再将当前奇数单独成组
        if (currentBuffer != null)
        {
            result.Add(currentBuffer);
            currentBuffer = null;
        }
        result.Add(new List<int> { num });
    }
    else
    {
        // 遇到偶数：初始化或追加到当前缓冲
        currentBuffer ??= new List<int>();
        currentBuffer.Add(num);
    }
}

// 处理最后剩余的偶数缓冲
if (currentBuffer != null)
{
    result.Add(currentBuffer);
}

方法二：Rx响应式实现 #

如果要基于Rx操作符实现，可以用Window配合自定义边界逻辑，精准控制分组的开始与结束：

var origin = new[] { 1, 3, 5, 7, 2, 4, 6, 8, 1 }.ToObservable();

var result = origin
    .Window(() => origin
        .Skip(1)
        .Where(x => 
            // 边界触发条件：下一个元素是奇数，或者下一个元素是偶数且当前元素是奇数
            x % 2 != 0 || (x % 2 == 0 && origin.Take(1).First() % 2 != 0)
        )
    )
    .SelectMany(window => window.ToList());

原代码失败原因 #

原代码中boundaries是奇偶状态变化的信号，但Buffer会在每次边界触发时结束当前缓冲并开启新缓冲，这会导致：

• 第一个奇数触发边界时，缓冲为空；第二个奇数触发时，缓冲里是前一个奇数，分组结果完全不符合预期
• 偶数起始的边界信号没有对应到正确的分组起始点，逻辑完全混乱

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