關(guān)于寫異步代碼測(cè)試用例的一些思考
如果說異步代碼不好寫是共識(shí)的話，那么寫異步代碼測(cè)試用例就更難了。最近我剛剛完成了一個(gè) Flaky 測(cè)試，所以想和大家分享一些關(guān)于寫異步測(cè)試用例的想法。
這篇文章里，我們會(huì)探索一個(gè)關(guān)于異步測(cè)試用例的常見問題 —— 如何強(qiáng)制規(guī)定某些線程的順序，如何強(qiáng)制某一個(gè)線程操作早于另一些執(zhí)行。通常我們并不想強(qiáng)行規(guī)定線程之間的順序，因?yàn)檫@違背了多線程的原則，所謂多線程就是 為了做到并發(fā)，從而使得 CPU 可以根據(jù)當(dāng)前資源及應(yīng)用狀態(tài)選擇最佳的執(zhí)行順序。但是在測(cè)試中，為了確保測(cè)試結(jié)果的穩(wěn)定性，又必須明確線程順序。
測(cè)試節(jié)流閥（Throttler）
在軟件業(yè)里節(jié)流閥指的是用于限制并發(fā)操作個(gè)數(shù)，預(yù)留資源的模式，好比連接池，網(wǎng)絡(luò)緩存，或者 CPU 密集型操作。和其他同步工具不同的是，節(jié)流閥的角色是啟動(dòng)“快速失敗”機(jī)制，即促使超額請(qǐng)求立即失敗，而不是等待。“快速失敗”機(jī)制之所以重要，是因?yàn)榍?換操作，等待操作會(huì)消耗資源 —— 端口，線程，內(nèi)存等。
以下就是一個(gè)節(jié)流閥的簡(jiǎn)單實(shí)現(xiàn)（基本上是信號(hào)量的包裝，實(shí)際應(yīng)用中應(yīng)該是等待，重試等等）
- class ThrottledException extends RuntimeException("Throttled!")
- class Throttler(count: Int) {
- private val semaphore = new Semaphore(count)
- def apply(f: => Unit): Unit = {
- if (!semaphore.tryAcquire()) throw new ThrottledException
- try {
- f
- } finally {
- semaphore.release()
- }
- }
- }
現(xiàn)在我們開始基本的單元測(cè)試：測(cè)試單線程的節(jié)流閥（我們使用測(cè)試框架 specs2)。本例里，我們會(huì)驗(yàn)證順序調(diào)用是否會(huì)超過節(jié)流閥的最大限制（maxCount變量如下所示）。注意，這里我們用的是單線程，所以我們并不驗(yàn)證節(jié)流閥的“快速失敗”功能，這里的節(jié)流閥都處于不飽和狀態(tài)。事實(shí)上，我們只會(huì)測(cè)試節(jié)流閥在不飽和狀態(tài)下不會(huì)終止操作。
- class ThrottlerTest extends Specification {
- "Throttler" should {
- "execute sequential" in new ctx {
- var invocationCount = 0
- for (i <- 0 to maxCount) {
- throttler {
- invocationCount += 1
- }
- }
- invocationCount must be_==(maxCount + 1)
- }
- }
- trait ctx {
- val maxCount = 3
- val throttler = new Throttler(maxCount)
- }
- }
測(cè)試并發(fā)節(jié)流閥
前一個(gè)例子里，節(jié)流閥處于不飽和狀態(tài)，因?yàn)閱尉€程里節(jié)流閥一般都不會(huì)飽和。下面我們來測(cè)試一下多線程環(huán)境下節(jié)流閥是否還能工作良好。
設(shè)置如下：
- val e = Executors.newCachedThreadPool()
- implicit val ec: ExecutionContext=ExecutionContext.fromExecutor(e)
- private val waitForeverLatch = new CountDownLatch(1)
- override def after: Any = {
- waitForeverLatch.countDown()
- e.shutdownNow()
- }
- def waitForever(): Unit = try {
- waitForeverLatch.await()
- } catch {
- case _: InterruptedException =>
- case ex: Throwable => throw ex
- }
ExecutionContext 用來構(gòu)建 Future，waitForever 方法用來持有線程，直到測(cè)試結(jié)束前的鎖釋放。接下來的函數(shù)里，我們會(huì)關(guān)閉一個(gè)執(zhí)行服務(wù)。
以下就是一個(gè)測(cè)試節(jié)流器多線程行為的例子：
- "throw exception once reached the limit [naive,flaky]" in new ctx {
- for (i <- 1 to maxCount) {
- Future {
- throttler(waitForever())
- }
- }
- throttler {} must throwA[ThrottledException]
我們創(chuàng)建了 maxCount 個(gè)線程（調(diào)用 Future{}）來調(diào)用 waitForever 函數(shù)，該函數(shù)會(huì)一直直到道測(cè)試結(jié)束。然后我們繞開節(jié)流閥執(zhí)行另一個(gè)操作 —— maxCount + 1。預(yù)期的行為是，此時(shí)應(yīng)該拋出 ThrottledException 例外。但是，也許預(yù)期的例外并不發(fā)生，因?yàn)榻恿ζ鞯淖詈蟮囊粋€(gè)調(diào)用可能會(huì)比 future 里的先執(zhí)行（future 里會(huì)拋出例外，但是這不是預(yù)期結(jié)果）。
上面這個(gè)測(cè)試的問題是，在像期望中那樣節(jié)流閥拋出異常然后導(dǎo)致節(jié)流閥被違反之前，我們無法確定所有的線程都已經(jīng)開始并且在 waitForever 函數(shù)中被阻塞。為了修復(fù)這個(gè)問題，我們需要一些方法去等待所有 future 開始。這有一個(gè)我們大多數(shù)都很熟悉的一種方法：只要增加一個(gè) sleep 函數(shù)等待一些合適的時(shí)間。
- "throw exception once reached the limit [naive, bad]" in new ctx {
- for (i <- 1 to maxCount) {
- Future {
- throttler(waitForever())
- }
- }
- Thread.sleep(1000)
- throttler {} must throwA[ThrottledException]
- }
好了，現(xiàn)在這個(gè)測(cè)試幾乎都能通過了，但是這個(gè)方法還是錯(cuò)的因?yàn)橄旅孢@兩個(gè)原因：
測(cè)試持續(xù)的時(shí)間至少會(huì)和我們?cè)O(shè)置好的”合適的時(shí)間”差不多久。
在非常罕見的情況下，比如機(jī)器處于高負(fù)載的時(shí)候，這個(gè)合適的時(shí)間不一定足夠。
如果你仍然感到疑惑，可以搜索一下 Google 更多的原因。
一個(gè)更好的方式是將我們的線程(future)的開始和我們期望的東西同步起來。我們來使用 java.util.concurrent 里面的 CountDownLatch 類：
- "throw exception once reached the limit [working]" in new ctx {
- val barrier = new CountDownLatch(maxCount)
- for (i <- 1 to maxCount) {
- Future {
- throttler {
- barrier.countDown()
- waitForever()
- }
- }
- }
- barrier.await(5, TimeUnit.SECONDS) must beTrue
- throttler {} must throwA[ThrottledException]
- }
我們使用 CountDownLatch 處理障礙同步。 這個(gè)等待的方法會(huì)阻塞主線程直到鎖存計(jì)數(shù)變?yōu)?0。隨著其它線程的運(yùn)行（我們把這些其它線程表示為 future），每一個(gè) future 都會(huì)調(diào)用 countDown 方法使鎖存計(jì)數(shù)減 1。一但計(jì)數(shù)變?yōu)?0，所有的 future 就都已經(jīng)運(yùn)行到 waitForever 方法中了。
通過那一點(diǎn)，我們可以確保 throttler 是飽和的，內(nèi)部有最大數(shù)量（maxCount）的線程。另一個(gè)線程試圖進(jìn)入 throttler 將導(dǎo)致異常。我們有一個(gè)確定的方式建立我們的測(cè)試，測(cè)試會(huì)有一個(gè)主線程進(jìn)入 throttler。主線程可以恢復(fù)到這個(gè)點(diǎn)（門閂計(jì)數(shù)為 0 并等 CountDownLatch 釋放等待線程）。
如果一些意想不到的事情發(fā)生，我們使用超時(shí)略高保障避免無限阻塞發(fā)生。如果這樣的事情發(fā)生，我們的測(cè)試就失敗了。這個(gè)超時(shí)不會(huì)影響到測(cè)試時(shí)間，除非發(fā)生意外情況，否則，我們都不應(yīng)該等待。
結(jié)論
測(cè)試異步程序時(shí)，通常需要在具體的測(cè)試用例中指定多個(gè)線程之間的執(zhí)行順序。不使用任何同步策略的測(cè)試是不可靠的，測(cè)試結(jié)果有時(shí)成功有時(shí)失敗。使用 Thread.sleep 降低了測(cè)試出錯(cuò)的概率，但沒有完全解決這個(gè)問題。
在大多數(shù)情況下，當(dāng)需要在測(cè)試中保證多個(gè)線程的執(zhí)行順序時(shí)，可以使用 CountDownLatch 代替 Thead.sleep。使用 CountDownlatch 的好處是通過它可以指定釋放（保持）線程的時(shí)機(jī)，有兩個(gè)優(yōu)點(diǎn)：確保按順序執(zhí)行使測(cè)試結(jié)果更可靠；加快了測(cè)試程序的執(zhí)行速度。即使對(duì)于普通的 waiting 操作，比如 waitForever 函數(shù)，盡管也可以使用 Thread.sleep(Long.MAX_VALUE) 這樣的函數(shù)實(shí)現(xiàn)，但為了保證程序的健壯性最好不要這樣做。
完整的代碼可以在 GitHub 中找到。
