Rx.NET如何实现设置容量、满时丢弃最旧元素的功能

Rx.NET没有自带和Channel完全同名的配置API，但完全可以用内置能力实现和你给出的代码一致的效果：指定队列容量、满时丢弃最早未处理项、处理消息耗时较长时仅保留最新待处理项，中间到达的消息全部丢弃。

先理清楚你贴的BoundedChannelOptions(1) + DropOldest配置的实际运行逻辑，和你描述的场景完全对齐：

• 内部队列最多缓存1条还没被消费的消息
• 消息被消费者取出开始处理后，队列直接腾空
• 处理消息的这段时间里新到的消息，第一条会先进队列，后续再到的新消息会直接把队列里已存的旧消息挤掉，永远保证队列里最多只有1条、且是最新的待处理消息
• 当前消息处理完，消费者会立刻取出队列里存的最新消息开始处理，没有待处理消息时就等待新消息到达

具体实现 #

不需要引入第三方依赖，直接结合SemaphoreSlim做并发控制，配合队列缓存逻辑就能复现完整行为，写成扩展方法可以直接复用：

public static IObservable<T> BoundedDropOldest<T>(this IObservable<T> source, int capacity = 1)
{
    if (capacity < 1) throw new ArgumentOutOfRangeException(nameof(capacity));
    return Observable.Create<T>(async (observer, cancellationToken) =>
    {
        var semaphore = new SemaphoreSlim(1, 1);
        var queue = new Queue<T>(capacity);
        var queueLock = new object();

        var subscription = source.Subscribe(
            onNext: item =>
            {
                lock (queueLock)
                {
                    // 队列满了就把最早的项出队丢弃
                    if (queue.Count >= capacity)
                        queue.Dequeue();
                    queue.Enqueue(item);
                }

                // 新消息入队后尝试启动消费，已经在消费中就直接返回
                if (semaphore.Wait(0, cancellationToken))
                {
                    _ = Task.Run(async () =>
                    {
                        try
                        {
                            while (!cancellationToken.IsCancellationRequested)
                            {
                                T current;
                                lock (queueLock)
                                {
                                    if (queue.Count == 0) break;
                                    current = queue.Dequeue();
                                }
                                observer.OnNext(current);
                            }
                        }
                        catch (Exception ex)
                        {
                            observer.OnError(ex);
                        }
                        finally
                        {
                            semaphore.Release();
                        }
                    }, cancellationToken);
                }
            },
            onError: observer.OnError,
            onCompleted: () =>
            {
                // 等队列里剩余的最后一项处理完再触发完成通知
                semaphore.Wait(cancellationToken);
                observer.OnCompleted();
            });

        return subscription;
    });
}

这个实现支持自定义容量，把capacity设为1就是你示例里的效果。

使用方式 #

和普通Rx算子用法一致，对应你提到的单条消息处理耗时10秒的场景示例：

// 模拟每秒产生1条新消息的源
var source = Observable.Interval(TimeSpan.FromSeconds(1));

source.BoundedDropOldest(capacity: 1)
    .Select(async item =>
    {
        await Task.Delay(10000); // 模拟单条消息10秒处理耗时
        Console.WriteLine($"已处理消息: {item}");
    })
    .Concat() // 保证处理逻辑串行执行，不并发
    .Subscribe();

运行后每10秒只会处理这10秒内到达的最新1条消息，中间的所有消息都会被丢弃，和你贴的Channel代码效果完全一致。

额外场景适配 #

如果你需要的是严格的处理期全丢逻辑——也就是只要当前有消息在处理，不管来多少新消息直接全丢，等当前处理完再接收下一条，完全不缓存待处理项，只需要把队列逻辑去掉，用标记位判断处理状态即可：

public static IObservable<T> DropAllWhileProcessing<T>(this IObservable<T> source, Func<T, Task> handler)
{
    return Observable.Create<T>(observer =>
    {
        int processingFlag = 0;
        return source.Subscribe(
            async item =>
            {
                // 标记位为1说明正在处理，直接丢弃新消息
                if (Interlocked.CompareExchange(ref processingFlag, 1, 0) != 0)
                    return;
                try
                {
                    await handler(item);
                    observer.OnNext(item);
                }
                catch (Exception ex)
                {
                    observer.OnError(ex);
                }
                finally
                {
                    Interlocked.Exchange(ref processingFlag, 0);
                }
            },
            observer.OnError,
            observer.OnCompleted);
    });
}

注意不要直接用默认的ObserveOn算子做背压，它内部的调度队列是无界的，会把所有上游消息全部缓存下来造成内存堆积，达不到丢弃旧消息的效果。

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