如何在单元测试中进行并发测试

发表于 | 分类于 |

摘要：本文记录一下如何在单元测试中进行并发测试。本文主要介绍3种方法，其中一个是使用第三方的依赖，另外2个是 Spockframework 提供的工具。

有时候我们需要在单元测试中测试并发，主要的场景有如下几种

  1. 测试的目标启动了新的线程，同时有回调函数，需要在单元测试中测试其回调函数是否执行。
  2. 测试中启动了多个线程
  3. 测试需要等待异步操作的结果

如果有如上需求的单元测试，需要引入并发测试的支持。

concurrentunit.Waiter

这个第三方库设计的很简洁，功能比较全面，其用法十分简单，只需要在主线程通过 waiter.await(10L,2) 阻塞等待子线程和回调，在回调或子线程中通过 waiter.resume() 取现主线程的阻塞。

其中 waiter.await(time,count) 第一个参数是等待的时间，也就是说超过这个时间还没有子线程或回调调用 resume 则会报超时的异常。第二个参数是等待 resume的个数，比如我们需要测试一个任务结束后，需要回到两个函数，那么我们就需要接受2个 resume 才能让主线程继续，否则还是会超时错误。

案例

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
def "test Post data with adding task listeners"() {
    when: "when"
    final Waiter waiter = new Waiter();
    Task task = Firebase.post("agent/xxxx1","NEWACTION")
    boolean isSuccess = false
    boolean isFailure = false
    boolean isComplete = false
    boolean isCompleteCheck = false
    task.addOnSuccessListener(new OnSuccessListener() {
        @Override
        void onSuccess(Object o) {
            isSuccess = true
            println("task is success.")
            waiter.resume();
        }
    })
    task.addOnFailureListener(new OnFailureListener() {
        @Override
        void onFailure(@NonNull Exception e) {
            isFailure = true
            println("task is failure." + e.getMessage())
            waiter.resume();
        }
    })
    task.addOnCompleteListener(new OnCompleteListener() {
        @Override
        void onComplete(@NonNull Task task1) {
            println("task is complete.")
            isComplete = true
            isCompleteCheck = task1.isSuccessful()
            waiter.resume();
        }
    })
    then: "then there are 2 callback will invoke"
    waiter.await(10 as long,2);
    task
    isSuccess
    !isFailure
    isComplete
    isCompleteCheck
}

BlockingVariable

这个是 spockframework 内置的工具类，位于spock.util.concurrent包中。

Class BlockingVariable

T - the variable’s type

A statically typed variable whose get() method will block until some other thread has set a value with the set() method, or a timeout expires. Useful for verifying state in an expect- or then-block that has been captured in some other thread.

通过名字我们可以猜出，该方法通过泛型将我们的业务类包装成一个可以被阻塞的变量。也就是说这个变量可以被阻塞，直到被赋值。通俗的讲就是如果我希望在某个子进程或者回调中向某个变量赋值，这时候我们就可以用这个BlockingVariable 将该变量进行包装，然后在主线程通过 xxx.get() 操作阻塞直到 xxx.set() 操作被执行。同样该方法也可以设置超时时间和时间单位：

BlockingVariable(int timeout, TimeUnit unit)

案例

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
def "test Post data with adding task listeners by using BlockingVariable"() {
    when: "when"
    def isSuccess = new BlockingVariable<Boolean>(10 as double)
    def isFailure = new BlockingVariable<Boolean>(10 as double)
    def isComplete = new BlockingVariable<Boolean>(10 as double)
    def isCompleteCheck = new BlockingVariable<Boolean>(10 as double)
    Task task = Firebase.post("agent/xxxx1","NEWACTION")
    task.addOnSuccessListener(new OnSuccessListener() {
        @Override
        void onSuccess(Object o) {
            isSuccess.set(true)
            println("task is success.")
        }
    })
    task.addOnFailureListener(new OnFailureListener() {
        @Override
        void onFailure(@NonNull Exception e) {
            println("task is failure." + e.getMessage())
        }
    })
    task.addOnCompleteListener(new OnCompleteListener() {
        @Override
        void onComplete(@NonNull Task task1) {
            println("task is complete.")
            isComplete.set(true)
            isCompleteCheck.set(task1.isSuccessful())
        }
    })
    then: "then"
    task
    isSuccess.get()
    isComplete.get()
    isCompleteCheck.get()
}

AsyncConditions

上面的spockframework提到的方法比较简单，一般是针对一个变量来测试。AsyncConditions 提供更多个测试空间。

Class AsyncConditions

Alternative to class BlockingVariable(s) that allows to evaluate conditions in a thread other than the spec runner’s thread(s). Rather than transferring state to an expect- or then-block, it is verified right where it is captured. On the upside, this can result in a more informative stack trace if the evaluation of a condition fails. On the downside, the coordination between threads is more explicit (number of evaluate blocks has to be specified if greater than one), and the usual structure of a feature method cannot be preserved (conditions are no longer located in expect-/then-block).

该工具有点类似于第一种方法，首先定义一个 AsyncConditions 对象，然后在主线程 conds.await() 阻塞，最后在子线程进行 conds.evaluate 来进行测试目标变量是否达到预期。同样在await 的时候可以设置超时时间：

await(double seconds)

案例

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
def "test Post data with adding task listeners by using AsyncConditions"() {
        when: "when"
        def conds = new AsyncConditions()
        Task task = Firebase.post("agent/xxxx1","XXX")
        task.addOnCompleteListener(new OnCompleteListener() {
            @Override
            void onComplete(@NonNull Task task1) {
                conds.evaluate {
                    println("task is complete.")
                    task1.isComplete()
                    task1.isSuccessful()
                }
            }
        })

        task.addOnSuccessListener(new OnSuccessListener() {
            @Override
            void onSuccess(Object o) {
                println("task is success.")
            }
        })

        task.addOnFailureListener(new OnFailureListener() {
            @Override
            void onFailure(@NonNull Exception e) {
                conds.evaluate {
                    println("task is failure." + e.getMessage())
                }
            }
        })
        then: "then"
        conds.await(1000 as Double)
    }