635
非阻塞同步算法實戰（一）
前言
本文寫給對ConcurrentLinkedQueue的實現和非阻塞同步算法的實現原理有一定了解，但缺少實踐經驗的朋友，文中包括了實戰中的嚐試、所走的彎路，經驗和教訓。
背景介紹
上個月，我被安排獨自負責一個聊天係統的服務端，因為一些原因，我沒使用現成的開源框架，網絡那塊直接使用AIO，收數據時，因為隻會從 channel裏過來，所以不需要考慮同步問題；但是發送數據時，因為有聊天消息的轉發，所以必需處理這個同步問題。AIO中，是處理完一個注冊的操作 後，再執行我們定義的方法，此時，如果還有數據需要寫，則繼續注冊寫操作，如果沒有則不注冊；提交數據時，如果當前沒有注冊寫操作，則注冊一個，否則僅提 交（此時再注冊則會報異常）。這樣，需要同步的點就是：如何知道當前還有沒有數據需要發送（因為其它線程也可以提交數據到此處），和如何知道此次提交時， 有沒有注冊寫操作。總之，要完成：有數據要發送時，必需有寫操作被注冊，並且隻能注冊一次；沒有數據時，不能有寫操作被注冊。
問題分析
經過分析，上麵的問題，可以抽象成：我需要知道當往隊列裏插入一條數據之前，該隊列是否為空，如果為空則應該注冊新的寫操作。當從隊列裏取出一條數據後，該隊列是否為非空，如果非空則應該繼續注冊寫操作。（本文之後以“關注的操作”來表示這種場景下的插入或取出操作）
目前的問題是，我使用的隊列是ConcurrentLinkedQueue，但是它的取出數據的方法，沒有返回值告訴我們從隊列裏取出數據之後隊列 是否為空，如果是使用size或peek方法來執行判斷，那就必需加鎖了，否則在拿到隊列大小時，可能隊列大小已經變化了。所以我首先想到的是，如何對該 隊列進行改造，讓它提供該信息。
注意：這裏指的不是當次而是之後，所以如果我們使用隊列的peek()方法返回null，就知道隊列是否為空，但是不知道之後是否為空 ，並且，當關注的操作發生時，在插入或取出操作的返回值裏告知此信息，來指導是否繼續注冊寫操作。
用鎖實現
如果使用鎖的話很容易處理，用鎖同步插入和取出方法，下麵是鎖實現的參考：
public E poll() {
synchronized (this) {
E re = q.poll();
// 獲取元素後，隊列是空，表示是我關注的操作
if (q.peek() == null) {
}
return re;
}
}
public void offer(E e) {
synchronized (this) {
// 插入元素前，隊列是空，表示是我關注的操作
if (q.peek() == null) {
}
q.offer(e);
}
}
但因為是服務端，我想用非阻塞同步算法來實現。
嚐試方案一
我第一次想到的改造辦法是，將head占位節點改成固定的，頭節點移動時，隻將head節點的next指向新的節點，在插入數據時，如果是在 head節點上成功執行的該操作，那麼該插入就是關注的的操作；在取出時，如果將head節點的next置為了null，那麼該取出就是關注的操作（ 因 為之前的占位節點是變化的，所以沒法比較，必需用同步，現在是固定的了，所以可以直接與head節點進行比較 ）。如此一來，問題好像被解決了。改造完之 後，出於嚴謹，我仔細讀了一遍代碼，發現引入了新的問題，我的取出操作是這樣寫的
/**
* @author trytocatch@163.com
*/
public E poll(){
for(;;){
Node n = head.nextRef.get();//head指向固定的head節點，為final
if(n == null)
return null;
Node m = n.nextRef.get();
if(head.next.compareAndSet(n,m){
if(m==null)
;//此時為關注的操作（為了簡化代碼顯示，不將該信息當作返回值返回了，僅注釋）
return n.itemRef.get();
}
}
}
這裏有一個致命的問題：如果m為null，在CAS期間，插入了新節點，n的next由null變成了非null，緊接著又把head的next更 新為了null，那麼鏈就斷了，該方法還存在一些其它的問題，如當隊列為空的時候，尾節點指向了錯誤的位置，本應該是head的。我認為最根本的原因在 於，head不能設為固定的，否則會引發一堆問題。第一次嚐試宣告失敗。
嚐試方案二
這次我嚐試將head跟tail兩個引用包裝成一個對象，然後對這個對象進行CAS更新（這僅為一處理論上的嚐試，因為從性能上麵來講，已經大打折 扣了，創建了很多額外的對象），如果head跟tail指向了同一個節點，則認為隊列是空的，根據此信息來判斷一個操作是不是關注的操作。但該嚐試僅停留 在了理論分析階段，期間發現了一些小問題，沒法解決，後來我發現，我把ConcurrentLinkedQueue原本可以分成兩步執行的插入和取出操作 （更新節點的next或item引用，然後再更新head或tail引用），變成了必需一步完成，ConcurrentLinkedQueue尚不能一步 完成，我何德何能，可將它們一步完成？所以直接放棄了。
解決方案一
經過兩次的失敗嚐試，我幾乎絕望了，我懷疑這是不是不能判斷出是否為關注的操作。
因為是在做項目，周末已經過去了，不能再搞這些“研究”了，所以我退而求其次，想了個不太漂亮的辦法，在隊列外部維護一個變量，記錄隊列有多大，在 插入或取出後，更新該變量，使用的是AtomicInteger，如果更新時，將變量從1變成0，或從0變成了1，就認為該插入或取出為關注的操作。
private AtomicInteger size = new AtomicInteger(0);
public E poll() {
E re = q.poll();
if (re == null)
return null;
for(int old;;){
old = size.get();
if(size.compareAndSet(old,old-1)){
// 獲取元素後，隊列是空，表示是我關注的操作
if(old == 1){
}
break;
}
}
return re;
}
public void offer(E e) {
q.offer(e);
for(int old;;){
old = size.get();
if(size.compareAndSet(old,old+1)){
// 插入元素前，隊列是空，表示是我關注的操作
if(old == 0){
}
break;
}
}
}
此時，也許細心的朋友會問，因為沒有使用鎖，這個變量並不能真實反映隊列的大小，也就不能確定它是不是關注的操作。沒錯，是不能真實反映，但是，我 獲取關注的操作的目的是用來指導我：該不該注冊新的寫操作，該變量的值變化就能提供正確的指導，所以，同樣是正確的，隻不過途徑不同而已。理論上的分析和 後來的項目正確運行都印證了該方法的正確性。
解決方案二
因為上麵的方法額外加了一次lock-free級別的CAS操作，我心裏總不太舒服，空餘時間總在琢磨，真的就沒有辦法，在不增加額外lock-free級別CAS開支的情況下，知曉一個操作是不是關注的操作？
後來經分析，如果要知曉是不是關注的操作，跟兩個數據有關，真實的頭節點跟尾節點（不同於head跟tail，因為它們是滯後的，之前將它們包裝成 一個對象就是犯了該錯誤），ConcurrentLinkedQueue的實現中，這兩個節點是沒有競爭的，一個是更新item,一個是更新next，必 需得讓他們競爭同一個東西，才能解決該問題，於是我想到了一個辦法，取出完成後，如果該節點的next為null，就將其用CAS置為一個特殊的值，若成 功則認為是關注的操作；插入成功後，如果next被替換掉的值不是null而是這個特殊值，那麼該插入也為關注的操作。這僅增加了一次wait-free 級別的CAS操作（取出後的那次CAS），perfect!
因為ConcurrentLinkedQueue的很多變量、內部類都是私有的，沒法通過繼承來改造，沒辦法，隻得自行實現。對於隊列裏使用的 Node，實現的方式有很多種，可以使用AtomicReference、AtomicReferenceFieldUpdater來實現，如果你願意的 話，甚至是像ConcurrentLinkedQueue一樣，用sun.misc.Unsafe來實現（注意：一般來說，sun包下的類是不推薦使用 的），各有優缺點吧，所以我就不提供該隊列的具體實現了，下麵給出在ConcurrentLinkedQueue（版本：1.7.0_10）基礎上進行的 改造，供參考。注意，如果需要用到size等方法，因為特殊值的引入，影響了之前的判斷邏輯，應重新編寫。
/**
* @author trytocatch@163.com
*/
private static final Node MARK_NODE = new Node(null);
public boolean offer(E e) {
checkNotNull(e);
final Node newNode = new Node(e);
for (Node t = tail, p = t;;) {
Node q = p.next;
if (q == null || q == MARK_NODE) {//修改1：加入條件：或q == MARK_NODE
// p is last node
if (p.casNext(q, newNode)) {//修改2：將null改為q
// Successful CAS is the linearization point
// for e to become an element of this queue,
// and for newNode to become "live".
if (q == MARK_NODE)//修改3:
;//此時為關注的操作（為了簡化代碼顯示，僅注釋）
if (p != t) // hop two nodes at a time
casTail(t, newNode); // Failure is OK.
return true;
}
// Lost CAS race to another thread; re-read next
}
else if (p == q)
// We have fallen off list. If tail is unchanged, it
// will also be off-list, in which case we need to
// jump to head, from which all live nodes are always
// reachable. Else the new tail is a better bet.
p = (t != (t = tail)) ? t : head;
else
// Check for tail updates after two hops.
p = (p != t && t != (t = tail)) ? t : q;
}
}
public E poll() {
restartFromHead:
for (;;) {
for (Node h = head, p = h, q;;) {
E item = p.item;
if (item != null && p.casItem(item, null)) {
// Successful CAS is the linearization point
// for item to be removed from this queue.
if (p != h) // hop two nodes at a time
updateHead(h, ((q = p.next) != null) ? q : p);
if (p.casNext(null,MARK_NODE))//修改1：
;//此時為關注的操作
return item;
}
else if ((q = p.next) == null) {
updateHead(h, p);
return null;
}
else if (p == q)
continue restartFromHead;
else
p = q;
}
}
}
小結
設計非阻塞算法的關鍵在於，找出競爭點，如果獲取的某個信息跟兩個操作有關，那麼應該讓這兩個操作競爭同一個東西，這樣才能反應出它們的關係。
文章轉自 並發編程網-ifeve.com
最後更新：2017-05-22 16:02:36