如何在发生事件B时取消已防抖的Rx事件A？

我需要实现这样的逻辑：当事件B发生时，取消正在防抖处理的事件A。比如，若期间点击了鼠标按钮，就忽略已防抖的键盘按键事件。

我用计时器延迟模拟事件A和B，尝试用Rx.NET的Throttle操作符对A进行防抖，但结果不符合预期：

var subjA = new Subject<int>();
var subjB = new Subject<Unit>();

// 期望输出：3（因为B在时间线150ms时触发）
// 实际输出：2, 3

subjA
    .Throttle(TimeSpan.FromMilliseconds(200))
    .Subscribe(s => Console.WriteLine(s)); 

await Task.WhenAll(EmitA(), EmitB(), Task.Delay(2000));

async Task EmitA()
{
    subjA!.OnNext(1);
    await Task.Delay(100);
    subjA!.OnNext(2);
    await Task.Delay(500);
    subjA!.OnNext(3);
}

async Task EmitB()
{
    await Task.Delay(150);
    subjB!.OnNext(Unit.Default);
}

我已经通过Select/Delay/TakeUntil/Switch的组合解决了问题，但感觉这个写法对于Rx的常见场景来说过于复杂：

#nullable enable
using System.Reactive;
using System.Reactive.Linq;
using System.Reactive.Subjects;
using System.Threading.Tasks;
using System;

var subjA = new Subject<int>();
var subjB = new Subject<Unit>();

subjA.Select(n =>
        Observable.Return(n).Delay(TimeSpan.FromMilliseconds(200))
        .TakeUntil(subjB))
    .Switch()
    .Subscribe(s => Console.WriteLine(s));

await Task.WhenAll(EmitA(), EmitB(), Task.Delay(2000));

async Task EmitA()
{
    subjA!.OnNext(1);
    await Task.Delay(100);
    subjA!.OnNext(2);
    await Task.Delay(500);
    subjA!.OnNext(3);
}

async Task EmitB()
{
    await Task.Delay(150);
    subjB!.OnNext(Unit.Default);
}

有没有更优雅的实现方式？

大理石图：

subjA:  +---1---2---------------3---------|
subjB:  +----------U-------------------------------|
Result: +-----------------------------3---|

你可以用Observable.Timer替代Return+Delay，让代码更简洁，核心逻辑和你的方案一致，但写法更紧凑：

subjA
    .Select(n => Observable.Timer(TimeSpan.FromMilliseconds(200))
        .Select(_ => n)
        .TakeUntil(subjB))
    .Switch()
    .Subscribe(Console.WriteLine);

如果这类场景在你的代码中频繁出现，还可以封装成自定义操作符，进一步提升复用性和代码可读性：

public static class ObservableExtensions
{
    public static IObservable<T> ThrottleUntil<T, TInterrupt>(
        this IObservable<T> source,
        TimeSpan dueTime,
        IObservable<TInterrupt> interruptSignal)
    {
        return source
            .Select(item => Observable.Timer(dueTime)
                .Select(_ => item)
                .TakeUntil(interruptSignal))
            .Switch();
    }
}

使用时就非常简洁：

subjA
    .ThrottleUntil(TimeSpan.FromMilliseconds(200), subjB)
    .Subscribe(Console.WriteLine);

这个自定义操作符清晰表达了“防抖直到中断信号触发”的逻辑，完全贴合你的需求，同时保持了Rx代码的声明式风格。

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