如何在xUnit中同步并发运行的测试？附NUnit实现求更佳方案

我需要并行运行数十个测试用例，但有个特殊要求：所有测试运行到某个节点时，有一个操作仅需执行一次（或限流执行，避免100个并发触发100次），操作完成后测试再继续独立运行。

目前我在NUnit里的实现如下：

[TestFixture] 
public class ScheduleJobTest { 
    private static readonly Subject<Unit> s_tickets = new Subject<Unit>(); 

    [OneTimeSetUp] 
    public static void SetUp() { 
        Task.Run(() => s_tickets 
            .Throttle(TimeSpan.FromSeconds(1)) 
            .SelectMany(_ => PollSchedulesNow()) 
            .DefaultIfEmpty() 
            .GetAwaiter() 
            .GetResult()); 
    } 

    [OneTimeTearDown] 
    public static void TearDown() { 
        s_tickets.OnCompleted(); 
    } 

    private static IObservable<Unit> PollSchedulesNow() { 
        // Do something that eventually unblocks all the test cases 
    } 

    [TestCaseSource(nameof(ProvideNamespaces)), Parallelizable(ParallelScope.Children)] 
    public void ScheduleJob(NamespaceConnectionInfo nci) { 
        // Do something 
        s_tickets.OnNext(Unit.Default); 
        // Resume - blocks until the actions of PollSchedulesNow take effect 
    } 

    private static IEnumerable<TestCaseData> ProvideNamespaces() { 
        return ... 
    } 
}

这个实现里，每个测试用例给特殊操作排队一个ticket，限流机制确保1秒内到达的ticket不会重复触发操作，停顿后才会触发（后续有测试的话可以再次执行，这是可接受的，我只是不想100个并发测试触发100次）。

现在想请教：在xUnit中有没有更合适的实现方式？或者是否需要更换其他测试框架？

xUnit没有NUnit里OneTimeSetUp那种静态级别的全局初始化，但可以通过下面几种方式完美实现你的需求：

1. 利用ICollectionFixture实现全局共享操作 #

xUnit的ICollectionFixture可以让多个测试类共享一个fixture实例，如果你所有目标测试用例都归属于同一个集合，这是最贴合你场景的方案：

首先定义一个共享fixture，把限流触发逻辑封装进去：

public class PollScheduleFixture : IDisposable
{
    private readonly Subject<Unit> _tickets = new Subject<Unit>();
    private readonly IDisposable _subscription;

    public PollScheduleFixture()
    {
        // 启动限流订阅逻辑，和你NUnit里的逻辑完全一致
        _subscription = _tickets
            .Throttle(TimeSpan.FromSeconds(1))
            .SelectMany(_ => PollSchedulesNow())
            .Subscribe();
    }

    private static IObservable<Unit> PollSchedulesNow()
    {
        // 你的核心操作逻辑，完成后解锁所有等待的测试
    }

    public void EnqueueTicket()
    {
        _tickets.OnNext(Unit.Default);
        // 如果需要阻塞直到操作完成，可以在这里补充信号量或Observable等待逻辑
    }

    public void Dispose()
    {
        _subscription.Dispose();
        _tickets.OnCompleted();
    }
}

然后给测试类标记对应的集合：

[CollectionDefinition("ScheduleJobCollection")]
public class ScheduleJobCollection : ICollectionFixture<PollScheduleFixture>
{
    // 这个类不需要实现任何逻辑，仅用于标记集合与fixture的关联
}

[Collection("ScheduleJobCollection")]
public class ScheduleJobTest
{
    private readonly PollScheduleFixture _fixture;

    public ScheduleJobTest(PollScheduleFixture fixture)
    {
        _fixture = fixture;
    }

    [Theory]
    [MemberData(nameof(ProvideNamespaces))]
    public void ScheduleJob(NamespaceConnectionInfo nci)
    {
        // 测试前置操作
        _fixture.EnqueueTicket();
        // 操作完成后继续执行测试逻辑
    }

    public static IEnumerable<object[]> ProvideNamespaces()
    {
        // 返回你的测试数据
        yield return new object[] { new NamespaceConnectionInfo(...) };
    }
}

这种方式的好处是fixture的生命周期和整个测试集合绑定，只会初始化一次，完美匹配你需要的全局限流逻辑。

2. 静态类+延迟订阅实现轻量化方案 #

如果不想用集合fixture，也可以用静态类结合延迟初始化来实现：

public static class PollScheduleCoordinator
{
    private static readonly Subject<Unit> _tickets = new Subject<Unit>();
    private static readonly Lazy<IDisposable> _subscription = new Lazy<IDisposable>(() =>
    {
        return _tickets
            .Throttle(TimeSpan.FromSeconds(1))
            .SelectMany(_ => PollSchedulesNow())
            .Subscribe();
    });

    static PollScheduleCoordinator()
    {
        // 确保测试结束时清理资源
        AppDomain.CurrentDomain.DomainUnload += (s, e) =>
        {
            _subscription.Value.Dispose();
            _tickets.OnCompleted();
        };
    }

    private static IObservable<Unit> PollSchedulesNow()
    {
        // 你的核心操作逻辑
    }

    public static void EnqueueTicket()
    {
        _subscription.Value; // 确保订阅已启动
        _tickets.OnNext(Unit.Default);
    }
}

测试里直接调用即可：

public class ScheduleJobTest
{
    [Theory]
    [MemberData(nameof(ProvideNamespaces))]
    public void ScheduleJob(NamespaceConnectionInfo nci)
    {
        // 测试前置操作
        PollScheduleCoordinator.EnqueueTicket();
        // 继续执行测试逻辑
    }

    public static IEnumerable<object[]> ProvideNamespaces()
    {
        // 返回你的测试数据
    }
}

这种方式更轻量化，但要注意资源清理的时机，避免内存泄漏。

3. 无Rx依赖的简单限流实现 #

如果你不需要Rx的复杂响应式逻辑，也可以用SemaphoreSlim结合Task.Delay实现更简洁的限流：

public static class PollScheduleCoordinator
{
    private static readonly SemaphoreSlim _semaphore = new SemaphoreSlim(1, 1);
    private static Task _pendingTask = Task.CompletedTask;

    public static async Task EnqueueAndWait()
    {
        await _semaphore.WaitAsync();
        try
        {
            if (_pendingTask.IsCompleted)
            {
                // 延迟1秒，等待更多测试加入后再执行操作
                _pendingTask = Task.Delay(TimeSpan.FromSeconds(1))
                    .ContinueWith(async _ =>
                    {
                        await PollSchedulesNow();
                    }, TaskScheduler.Default).Unwrap();
            }
            await _pendingTask;
        }
        finally
        {
            _semaphore.Release();
        }
    }

    private static async Task PollSchedulesNow()
    {
        // 你的核心操作逻辑
    }
}

测试里调用await PollScheduleCoordinator.EnqueueAndWait();即可，这种方式不需要引入Rx，代码更简洁，适合简单的限流场景。

其实xUnit完全能满足你的需求，上面几种方案都可以实现类似NUnit的效果，甚至更灵活。如果你的团队已经熟悉xUnit，完全没必要换框架。如果需要更复杂的全局生命周期管理，NUnit确实有更直接的API，但xUnit的fixture体系足够覆盖这个场景。

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