一个简单的测试模式可以用于一些(不是所有!)用例，就是多次重复相同的测试。例如，假设你有一个方法:

def process(input):
    # Spawns several threads to do the job
    # ...
    return output

创建一堆测试:

process(input1) -> expect to return output1
process(input2) -> expect to return output2
...

现在将每个测试运行多次。

如果流程的实现包含一个微小的错误(例如死锁、竞态条件等)，出现的概率为0.1%，那么运行1000次测试，则该错误至少出现一次的概率为64%。运行测试10000次，得到>99%的概率。

一个简单的测试模式可以用于一些(不是所有!)用例，就是多次重复相同的测试。例如，假设你有一个方法:

def process(input):
    # Spawns several threads to do the job
    # ...
    return output

创建一堆测试:

process(input1) -> expect to return output1
process(input2) -> expect to return output2
...

现在将每个测试运行多次。

如果流程的实现包含一个微小的错误(例如死锁、竞态条件等)，出现的概率为0.1%，那么运行1000次测试，则该错误至少出现一次的概率为64%。运行测试10000次，得到>99%的概率。

等待在帮助您编写确定性单元测试时也很有用。它允许您等待系统中的某个状态更新。例如:

await().untilCall( to(myService).myMethod(), greaterThan(3) );

or

await().atMost(5,SECONDS).until(fieldIn(myObject).ofType(int.class), equalTo(1));

它还支持Scala和Groovy。

await until { something() > 4 } // Scala example

上周我花了大部分时间在大学图书馆学习并发代码的调试。核心问题是并发代码是不确定的。通常，学术调试可以分为三个阵营之一:

Event-trace/replay. This requires an event monitor and then reviewing the events that were sent. In a UT framework, this would involve manually sending the events as part of a test, and then doing post-mortem reviews. Scriptable. This is where you interact with the running code with a set of triggers. "On x > foo, baz()". This could be interpreted into a UT framework where you have a run-time system triggering a given test on a certain condition. Interactive. This obviously won't work in an automatic testing situation. ;)

现在，正如上面评论者所注意到的，您可以将并发系统设计成更确定的状态。然而，如果你做得不好，你就又回到了设计顺序系统的问题上。

我的建议是，专注于制定一个非常严格的设计协议，规定什么是线程，什么不是线程。如果你限制了你的接口，使元素之间的依赖最小化，那就容易多了。

祝你好运，继续解决这个问题。

有一篇关于这个主题的文章，在示例代码中使用Rust作为语言:

https://medium.com/@polyglot_factotum/rust-concurrency-five-easy-pieces-871f1c62906a

总而言之，诀窍在于编写并发逻辑，使其对涉及多个执行线程的非确定性具有健壮性，使用通道和condvars等工具。

然后，如果这就是您构建“组件”的方式，那么测试它们的最简单方法是使用通道向它们发送消息，然后阻塞其他通道以断言组件发送某些预期的消息。

链接到的文章完全使用单元测试编写。

如果你正在测试简单的new Thread(runnable).run() 您可以模拟Thread来按顺序运行可运行对象

例如，如果被测试对象的代码像这样调用一个新线程

Class TestedClass {
    public void doAsychOp() {
       new Thread(new myRunnable()).start();
    }
}

然后模拟new Threads并按顺序运行runable参数会有所帮助

@Mock
private Thread threadMock;

@Test
public void myTest() throws Exception {
    PowerMockito.mockStatic(Thread.class);
    //when new thread is created execute runnable immediately 
    PowerMockito.whenNew(Thread.class).withAnyArguments().then(new Answer<Thread>() {
        @Override
        public Thread answer(InvocationOnMock invocation) throws Throwable {
            // immediately run the runnable
            Runnable runnable = invocation.getArgumentAt(0, Runnable.class);
            if(runnable != null) {
                runnable.run();
            }
            return threadMock;//return a mock so Thread.start() will do nothing         
        }
    }); 
    TestedClass testcls = new TestedClass()
    testcls.doAsychOp(); //will invoke myRunnable.run in current thread
    //.... check expected 
}