非阻塞同步演算法實戰（三）－LatestResultsProvider
感謝trytocatch投遞本文。
前言
閱讀本文前，需要讀者對happens-before比較熟悉，瞭解非阻塞同步的一些基本概念。本文主要為happens-before法則的靈活運用，和一些解決問題的小技巧，分析問題的方式。
背景介紹
原始需求為：本人當時在編寫一個正則替換工具，裡面會動態地顯示所有的匹配結果（包括替換預覽），文字、正則表示式、引數，這些資料的其中一項發生了變化，結果就應該被更新，為了提供友好的互動體驗，資料變化時，應該是發起一個非同步請求，由另一個獨立的執行緒來完成運算，完成後通知UI更新結果。由於是動態顯示，所以提交會非常頻繁。
需求描述
需要這樣一個工具類，允許使用者頻繁地提交資料（本文之後以“submit”表示該操作）和更新結果（本文之後以“update”表示該操作），submit時，如果當前有進行中的運算，則應該取消，使用新引數執行新的運算；update時，如果當前沒有進行中的運算（處於阻塞狀態），並且當前結果不是最新的，則喚醒該執行緒，使用當前的新資料，執行新的運算。此處之所以分為submit和update兩個方法，是為了支援手動更新，即點選更新按鈕時，才更新結果。
此外，出於練手的原因，也出於編寫一個功能全面，更實用的工具的目的，我還加入了一些額外的需求：
1、引入多執行緒場景，update和submit均可由多個執行緒同時發起，該工具類應設計成執行緒安全的。
2、允許延遲執行運算，如果延時內執行submit，僅重新計算延時。如果運算不方便取消，在短時間頻繁submit的場景下，延時會是一個很好的應對辦法。
3、允許設定一個最大延遲時間，作為延遲開啟運算的補充。當長時間頻繁submit時，會形成這樣的局面，一直未進入運算環節，新結果計算不出來，上一次計算結果卻是很早以前的。如果需要顯示一個較新但不是最新的結果，最大延遲時間將會很有用。
4、提供主動取消方法，主動取消正在進行的運算。
5、update時，允許等待運算完成，同時也可設定超時時間。當主動取消、超時、完成了當前或更(更加的意思)新的資料對應的運算時，結束等待。
需求交待完了，有興趣有精力的讀者，可以先試著思考下怎麼實現。
問題分析
該工具應該維護一個狀態欄位，這樣才能在發起某個操作時，根據所處的狀態作出正確的動作，如：如果當前不處於停止狀態（或者主動取消狀態，原因見下文），執行update就不需要喚醒運算執行緒。簡單分析可知，至少應該有這樣幾種狀態：
1、停止狀態：當前沒有運算任務，執行緒進入阻塞狀態，主動取消和運算完成後，進入該狀態
2、延遲狀態：設定了延遲開啟運算時，進入運算前，處於該狀態
3、運算狀態：正在執行運算
4、主動取消狀態：當發起主動取消時，進入該狀態
5、新任務狀態：當時有新的運算任務時，進入該狀態，然後重新進入運算狀態
延遲
再來看一下延遲，如果延遲500毫秒，就每次sleep(500)，那麼期間再submit怎麼辦？將它喚醒然後重新sleep(500)嗎？顯然不行，成本太大了。
我有一個小技巧：將500分成多個合適的等份，使用一個計數器，每次sleep一個等份，計數器加1，如果發起submit，僅把計數器置0即可，雖然看起來執行緒的狀態切換變多了，但應對頻繁重置時，它更穩定。雖然時間上會上下波動一個等份，但此處並不需要多麼精確。
現在還面臨這樣一個問題，如何知道當前是處於延遲狀態並計數器置0？取出狀態值進行判斷，然後置0，這方法顯然不行，因為置0的時候，可能狀態已經變了，所以你無法知道該操作是否生效了。
我想到的辦法是，再引入一個延遲重置狀態。如果處於該狀態，則下一次計數器加1時，將計數器重置，狀態變更是可以知道成功與否的。
狀態變更
有些狀態的變更是有條件的，比如說當前處於取消狀態，就不能把它轉為運算狀態，運算狀態只能由新任務狀態、延遲狀態（延遲完成後執行運算）或延遲重置狀態轉入。這種場景正好跟CAS一致，所以，使用一個AtomicInteger來表示狀態。
分析下各狀態之間的轉換，可以得出下面的狀態變更圖：
藍色的a(bcd)|(e)f線路為停止狀態下，發起一次update，運算完重新回到停止的過程，開啟延遲時是bcd，否則是e。
紅色的線j表示超過了最大延遲時間，退出延遲，進入運算狀態（也可以是d）。
綠色的線ghi（包括a）表示：如果發起了submit或update，狀態應該怎麼改變。如果處於延遲重置、新任務則不需要進行任何操作；如果處於延遲狀態，則轉為延遲重置即可；如果處於運算狀態，則可能使用了舊引數，應該轉為新任務；如果為主動取消或停止狀態，並且是呼叫update方法，則轉為新任務，並且可能處於阻塞狀態，應該喚醒該執行緒。
黑色的線l表示，可在任意狀態下發起主動取消，進入該狀態。然後通知等待執行緒後，轉入停止狀態，對應紫色的k，如果在停止狀態下發起主動取消，則僅轉為主動取消狀態，不會通知等待執行緒。所以當執行緒阻塞時，可能處於停止狀態或者主動取消狀態。
順序問題
上面已經分析到，當submit時，應該把延遲轉為延遲重置、或運算轉為新任務，這兩個嘗試的順序是不是也有講究呢？
是的，因為正常執行流程a(bcd)|(e)f中，運算狀態在延遲狀態之後，假如先嚐試運算轉為新任務，可能此時為延遲狀態，故失敗，再嘗試延遲轉為延遲重置時，狀態在這期間從剛才的延遲轉為了運算，故兩次嘗試都失敗了，本應該重置延遲的，卻什麼也沒幹，這是錯誤的。而將兩次嘗試順序調換一下，只要狀態為延遲或運算，那麼兩次狀態轉換嘗試中，一定有一次會成功。
之後的程式碼中還有多處類似的順序細節。
解決方案
下面給出完整的程式碼，除去等待運算完成那部分，其它地方均為wait-free級別的實現。
calculateResult是具體執行運算的方法；上文中的submit對應程式碼裡的updateParametersVersion方法，上文中的update對應剩餘幾個update方法。
updateAndWait方法中，使用了上一篇中講到的BoundlessCyclicBarrier，其維護的版本號就是引數的版本號ParametersVersion。
/** * @author [email protected] * @date 2013-2-2 */ public abstract class LatestResultsProvider { /** update return value */ public static final int UPDATE_FAILED = -1; public static final int UPDATE_NO_NEED_TO_UPDATE = 0; public static final int UPDATE_SUCCESS = 1; public static final int UPDATE_COMMITTED = 2; /** update return value */ /** work states*/ private static final int WS_OFF = 0; private static final int WS_NEW_TASK = 1; private static final int WS_WORKING = 2; private static final int WS_DELAYING = 3; private static final int WS_DELAY_RESET = 4; private static final int WS_CANCELED = 5; /** work states*/ private final AtomicInteger workState; private int sleepPeriod = 30; private final AtomicInteger parametersVersion; private volatile int updateDelay;// updateDelay>=0 private volatile int delayUpperLimit; private final BoundlessCyclicBarrier barrier; private Thread workThread; /** * * @param updateDelay unit: millisecond * @param delayUpperLimit limit the sum of the delay, disabled * while delayUpperLimit<0, unit: millisecond */ public LatestResultsProvider(int updateDelay, int delayUpperLimit) { if (updateDelay < 0) this.updateDelay = 0; else this.updateDelay = updateDelay; this.delayUpperLimit = delayUpperLimit; barrier = new BoundlessCyclicBarrier(0); workState = new AtomicInteger(WS_OFF); parametersVersion = new AtomicInteger(0); initThread(); } private void initThread() { workThread = new Thread("trytocatch's worker") { @Override public void run() { int sleepCount = 0; for (;;) { try { while (!workState.compareAndSet(WS_NEW_TASK, updateDelay > 0 ? WS_DELAY_RESET : WS_WORKING)) { if (workState.compareAndSet(WS_CANCELED, WS_OFF)) { barrier.cancel(); } LockSupport.park(); interrupted(); } if (workState.get() == WS_DELAY_RESET) { int delaySum = 0; for (;;) { if (workState.compareAndSet(WS_DELAY_RESET, WS_DELAYING)) { sleepCount = (updateDelay + sleepPeriod - 1) / sleepPeriod; } sleep(sleepPeriod); if (--sleepCount <= 0 && workState.compareAndSet(WS_DELAYING, WS_WORKING)) break; if (delayUpperLimit >= 0) { delaySum += sleepPeriod; if (delaySum >= delayUpperLimit) { if (!workState.compareAndSet( WS_DELAYING, WS_WORKING)) workState.compareAndSet( WS_DELAY_RESET, WS_WORKING); break; } } if (workState.get() != WS_DELAYING && workState.get() != WS_DELAY_RESET) break; } } if (isWorking()) { int workingVersion = parametersVersion.get(); try { calculateResult(); if (workState.compareAndSet(WS_WORKING, WS_OFF)) barrier.nextCycle(workingVersion); } catch (Throwable t) { t.printStackTrace(); workState.set(WS_CANCELED); } } } catch (InterruptedException e) { workState.compareAndSet(WS_DELAYING, WS_CANCELED); workState.compareAndSet(WS_DELAY_RESET, WS_CANCELED); } }// for(;;) }// run() }; workThread.setDaemon(true); workThread.start(); } public int getUpdateDelay() { return updateDelay; } /** * @param updateDelay * delay time. unit: millisecond */ public void setUpdateDelay(int updateDelay) { this.updateDelay = updateDelay < 0 ? 0 : updateDelay; } public int getDelayUpperLimit() { return delayUpperLimit; } /** * @param delayUpperLimit limit the sum of the delay, disabled * while delayUpperLimit<0, unit: millisecond */ public void setDelayUpperLimit(int delayUpperLimit) { this.delayUpperLimit = delayUpperLimit; } public final void stopCurrentWorking() { workState.set(WS_CANCELED); } /** * @return NO_NEED_TO_UPDATE, COMMITTED */ public final int update() { if (isResultUptodate()) return UPDATE_NO_NEED_TO_UPDATE; if (workState.compareAndSet(WS_CANCELED, WS_NEW_TASK) || workState.compareAndSet(WS_OFF, WS_NEW_TASK)) LockSupport.unpark(workThread); return UPDATE_COMMITTED; } /** * @param timeout * unit:nanoseconds * @return FAILED, NO_NEED_TO_UPDATE, SUCCESS * @throws InterruptedException */ public final int updateAndWait(long nanosTimeout) throws InterruptedException { int newVersion = parametersVersion.get(); if (update() == UPDATE_NO_NEED_TO_UPDATE) return UPDATE_NO_NEED_TO_UPDATE; barrier.awaitWithAssignedVersion(newVersion, nanosTimeout); return barrier.getVersion() - newVersion >= 0 ? UPDATE_SUCCESS : UPDATE_FAILED; } /** * @return FAILED, NO_NEED_TO_UPDATE, SUCCESS * @throws InterruptedException */ public final int updateAndWait() throws InterruptedException { return updateAndWait(0); } public final boolean isResultUptodate() { return parametersVersion.get() == barrier.getVersion(); } /** * be used in calculateResult() * @return true: the work state is working, worth to calculate the * result absolutely, otherwise you can cancel the current calculation */ protected final boolean isWorking() { return workState.get()==WS_WORKING; } /** * you must call this after update the parameters, and before calling the * update */ protected final void updateParametersVersion() { int pVersion = parametersVersion.get(); //CAS failed means that another thread do the same work already if (parametersVersion.compareAndSet(pVersion, pVersion + 1)) if (!workState.compareAndSet(WS_DELAYING, WS_DELAY_RESET)) workState.compareAndSet(WS_WORKING, WS_NEW_TASK); } /** * implement this to deal with you task */ protected abstract void calculateResult(); }
程式碼中，我直接在構造方法裡開啟了新的執行緒，一般來說，是不推薦這樣做的，但在此處，除非在構造還未完成時就執行update方法，否則不會引發什麼問題。
小結
狀態變更非常適合使用非阻塞演算法，並且還能夠達到wait-free級別。限於篇幅，有些沒講到的細節，請讀者藉助程式碼來理解吧，如有疑問，歡迎回復討論。
系列總結
本實戰系列就到此結束了，簡單總結下。
非阻塞同步相對於鎖同步而言，由程式碼塊，轉為了點，是另一種思考方式。
有時，無法做到一步完成，也許可以分成兩步完成，同樣可以解決問題，ConcurrentLinkedQueue就是這麼做的。
如果需要維護多個數據之間的某種一致關係，則可以將它們封裝到一個類中，更新時採用更新該類物件的引用的方式。
眾所周知，鎖同步演算法是難以測試的，非阻塞同步演算法更加難以測試，我個人認為，其正確性主要靠慎密的推敲和論證。
非阻塞同步演算法比鎖同步演算法要顯得更復雜些，如果對效能要求不高，對非阻塞演算法掌握得還不太熟練，建議不要使用非阻塞演算法，鎖同步演算法要簡潔得多，也更容易維護，如上面所說的，兩條看似沒有順序的語句，調換下順序，可能就會引發BUG。
