自旋鎖在高并發(fā)的用處，實(shí)現(xiàn)各種鎖

虛假喚醒：
由于莫名其妙的原因，線程有可能在沒有調(diào)用過notify()和notifyAll()的情況下醒來。這就是所謂的假喚醒（spurious wakeups）

.wikipedia的描述.png

我的demo測試，確實(shí)存在虛假喚醒，導(dǎo)致結(jié)果不一致，如圖將while自旋換成if判斷后 輸出的count值可能到不了10000，需要多測試幾遍。
我的理解，假如一個(gè)線程進(jìn)入if條件后進(jìn)行wait()釋放鎖，此時(shí)有別的線程在執(zhí)行++count,此時(shí)剛好發(fā)生虛喚醒（別問我怎么會發(fā)生，高并發(fā)就是這么巧，自己測試）那么就會執(zhí)行下面的語句，也進(jìn)行++count，其實(shí)相當(dāng)于兩個(gè)線程看到的比如都是77 ++后只變到了78，所以就導(dǎo)致了錯(cuò)誤的結(jié)果發(fā)生，自旋鎖while怎么避免呢，由于是while循環(huán)，即使被虛喚醒，那么該線程的代碼還是得執(zhí)行條件判斷，就又進(jìn)入了wait狀態(tài)（因?yàn)榧词拱l(fā)生虛喚醒事件，條件變量isLocked不可能變成false）所以解決了這個(gè)問題。但是自旋鎖是while循環(huán)，需要耗費(fèi)cpu資源的。


修改測試的地方

完整測試代碼

package com.alibaba.otter.canal.common;
import java.util.concurrent.CountDownLatch;

import org.junit.After;
import org.junit.Before;
import org.junit.Test;
@SuppressWarnings("restriction")
public class LockTest extends AbstractZkTest {
    private Object obj = new Object();
    private int count = 0;
    @Before
    public void setUp() {
        
    }

    @After
    public void tearDown() {
    }

    @Test
    public void testUnsafe() {
        CountDownLatch latch=new CountDownLatch(10000);
        for (int i = 0; i < 10000; i++) {
            Worker worker  = new Worker(latch);
            worker.start();
        }
        try {
            latch.await();
        } catch (InterruptedException e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        }
        
    }
    public int inc(){
        synchronized(this){
            System.out.println(count);
            return ++count;
            
        }
    }

    private Lock lock = new Lock();
    public int lockInc() {
        try {
            lock.lock();
        } catch (InterruptedException e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        }
        int newCount = ++count;
        System.out.println(count);
        lock.unlock();
        return newCount;
    }
    public class Lock{
        private boolean isLocked = false;

        public synchronized void lock()
            throws InterruptedException{
//          while(isLocked){
            if(isLocked){
                wait();
            }
            isLocked = true;
        }

        public synchronized void unlock(){
            isLocked = false;
            notify();
        }
    }
    class Worker extends Thread {
        private CountDownLatch latch;
        public Worker(CountDownLatch latch) {
            this.latch = latch;
        }
        public void run() {
//          inc();
            lockInc();
            latch.countDown();
        }
    }
}

實(shí)現(xiàn)可重入鎖
概念：可重入其實(shí)是同步代碼段中 發(fā)現(xiàn)是本線程 則不需要再wait，直接可以執(zhí)行 另一個(gè)同步代碼塊。java的synchronized同步是可重入的 例如

public synchronized outer(){
    inner();
}

public synchronized inner(){
    //do something
}

當(dāng)線程獲得鎖 進(jìn)入outer同步塊后 需要執(zhí)行inner 另一個(gè)同步塊，按理說此時(shí)是所有線程都去搶占inner代碼塊的鎖，但是可重入的話 獲得鎖的線程直接可以執(zhí)行inner語句
如果用lock代替synchronized的話一定要注意處理可重入性，避免死鎖。主要就是通過記錄是不是自己獲得了鎖，并且鎖了幾次，釋放鎖的時(shí)候?qū)?yīng)的將次數(shù)減少即可。這里附上完整的測試代碼

package com.alibaba.otter.canal.common;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;

import org.junit.After;
import org.junit.Before;
import org.junit.Test;
@SuppressWarnings("restriction")
public class LockTest extends AbstractZkTest {
    private Object obj = new Object();
    private int count = 0;
    @Before
    public void setUp() {
        
    }

    @After
    public void tearDown() {
    }

    @Test
    public void testUnsafe() {
        CountDownLatch latch=new CountDownLatch(10000);
        for (int i = 0; i < 10000; i++) {
            Worker worker  = new Worker(latch);
            worker.start();
        }
        try {
            latch.await();
        } catch (InterruptedException e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        }
        
    }
    public int inc(){
        synchronized(this){
            System.out.println(count);
            return ++count;
            
        }
    }

    private Lock lock = new Lock();
    public int lockInc() {
        int newCount=count;
        try {
            lock.lock();
            newCount = ++count;
            System.out.println(count);
        } catch (InterruptedException e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        } finally {
            lock.unlock();
        }
        
        return newCount;
    }
    public class Lock{
        private boolean isLocked = false;

        public synchronized void lock()
            throws InterruptedException{
            while(isLocked){
            
//          if(isLocked){
                
                wait();
            }
            isLocked = true;
        }

        public synchronized void unlock(){
            isLocked = false;
            notify();
        }
    }
    class Worker extends Thread {
        private CountDownLatch latch;
        public Worker(CountDownLatch latch) {
            this.latch = latch;
        }
        public void run() {
//          inc();
//          lockInc();
            try {
//              reentrantOuter();//可重入 synchronized方式
//              unReentrantOuter(); //lock未處理是否自己鎖 產(chǎn)生的是不可重入鎖，導(dǎo)致死鎖
                reentrantLockOuter();//lock方式實(shí)現(xiàn)可重入鎖
            } catch (Exception e) {
                // TODO Auto-generated catch block
                e.printStackTrace();
            }
            latch.countDown();
        }
    }
    

    /**
     * 
     * 可重入鎖的測試
     */
    @Test
    public void testReentrant() {
        CountDownLatch latch=new CountDownLatch(2);
        for (int i = 0; i < 2; i++) {
            Worker worker  = new Worker(latch);
            worker.start();
        }
        try {
            latch.await();
        } catch (InterruptedException e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        }
        
    }
    public synchronized void reentrantOuter() throws InterruptedException{
        System.out.println("reentrantOuter1");
        reentrantInner();
    }
    
    public synchronized void reentrantInner() throws InterruptedException{
        Thread.currentThread().sleep(10);
        System.out.println("reentrantInner2");
    }
    
    
    public void unReentrantOuter() throws InterruptedException{
        try {
            lock.lock();
            System.out.println("unReentrantouter1");
            unReentrantInner();
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            lock.unlock();
        }
    }

    public synchronized void unReentrantInner() {
        try {
            lock.lock();
            System.out.println("unReentrantInner2");
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            lock.unlock();
        }
    }
    
    public void reentrantLockOuter() {
        try {
            reentrantLock.lock();
            System.out.println("unReentrantouter1");
            reentrantLockInner();
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            reentrantLock.unlock();
        }
    }

    public synchronized void reentrantLockInner() throws InterruptedException{
        try {   
            reentrantLock.lock();
            System.out.println("unReentrantInner2");
            Thread.currentThread().sleep(10);
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            reentrantLock.unlock();
        }
    }
    
    ReentrantLock reentrantLock = new ReentrantLock();
    class ReentrantLock{
        boolean isLocked = false;
        Thread  lockedBy = null;
        int lockedCount = 0;

        public synchronized void lock()
            throws InterruptedException{
            Thread callingThread = Thread.currentThread();
            while(isLocked && lockedBy != callingThread){
                wait();
            }
            isLocked = true;
            lockedCount++;
            lockedBy = callingThread;
      }

        public synchronized void unlock(){
            if(Thread.currentThread() ==this.lockedBy){
                lockedCount--;
                if(lockedCount == 0){
                    isLocked = false;
                    notify();
                }
            }
        }
    }
}

上述鎖都是非公平的鎖，即先來的請求不一定是先處理，這樣的話就會導(dǎo)致有的線程可能很久得不到鎖（不要問為什么，并發(fā)大的話就是可能發(fā)生），這樣的話有些問題。我們基于此實(shí)現(xiàn)各公平的鎖。主要思路是 來的請求線程放到列表中，然后 notify的時(shí)候調(diào)用列表第一個(gè)的notify，即通知喚醒先來的請求線程即可。附上完整測試代碼。

package com.alibaba.otter.canal.common;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;

import org.junit.After;
import org.junit.Before;
import org.junit.Test;
@SuppressWarnings("restriction")
public class FairLockTest extends AbstractZkTest {
    private Object obj = new Object();
    private int count = 0;
    @Before
    public void setUp() {

    }

    @After
    public void tearDown() {
    }

    @Test
    public void testUnsafe() {
        CountDownLatch latch = new CountDownLatch(10000);
        for (int i = 0; i < 10000; i++) {
            Worker worker = new Worker(latch, i);
            worker.start();
        }
        try {
            latch.await();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

    }
    FairLock fairLock = new FairLock();
    public int fairLockInc() {
        int newCount = count;
        try {
            fairLock.lock();
            newCount = ++count;

        } catch (InterruptedException e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        } finally {
            fairLock.unlock();
        }

        return newCount;
    }

    class Worker extends Thread {
        private CountDownLatch latch;
        public Worker(CountDownLatch latch, int i) {
            this.latch = latch;
            this.setName(i + " thread");
        }
        public void run() {
            try {
                fairLockInc();
            } catch (Exception e) {
                e.printStackTrace();
            }
            latch.countDown();
        }
    }

    class FairLock {
        private boolean isLocked = false;
        private Thread lockingThread = null;
        private List<QueueObject> waitingThreads = new ArrayList<QueueObject>();

        public void lock() throws InterruptedException {
            QueueObject queueObject = new QueueObject();
            boolean isLockedForThisThread = true;
            synchronized (this) {
                System.out.println(Thread.currentThread().getName() + " in");
                waitingThreads.add(queueObject);
            }

            while (isLockedForThisThread) {
                synchronized (this) {
                    isLockedForThisThread = isLocked || waitingThreads.get(0) != queueObject;
                    if (!isLockedForThisThread) {
                        isLocked = true;
                        System.out.println(Thread.currentThread().getName()
                                + " out");
                        waitingThreads.remove(queueObject);
                        lockingThread = Thread.currentThread();
                        return;
                    }
                }
                try {
                    queueObject.doWait();
                } catch (InterruptedException e) {
                    synchronized (this) {
                        System.out.println(Thread.currentThread().getName()
                                + " out");
                        waitingThreads.remove(queueObject);
                    }
                    throw e;
                }
            }
        }

        public synchronized void unlock() {
            if (this.lockingThread != Thread.currentThread()) {
                throw new IllegalMonitorStateException(
                        "Calling thread has not locked this lock");
            }
            isLocked = false;
            lockingThread = null;
            if (waitingThreads.size() > 0) {
                waitingThreads.get(0).doNotify();
            }
        }
    }
    class QueueObject {

        private boolean isNotified = false;

        public synchronized void doWait() throws InterruptedException {

            while (!isNotified) {
                this.wait();
            }

            this.isNotified = false;
        }

        public synchronized void doNotify() {
            this.isNotified = true;
            this.notify();
        }

        public boolean equals(Object o) {
            return this == o;
        }

    }

}

其實(shí)我們基于此還可以實(shí)現(xiàn)更多的鎖，可以實(shí)現(xiàn)基于優(yōu)先級的鎖，主要實(shí)現(xiàn)思路就是創(chuàng)建線程的時(shí)候傳入優(yōu)先級參數(shù)，然后我們可以在入等待列表的時(shí)候?qū)Ρ葌魅氲膬?yōu)先級參數(shù)進(jìn)行比較大小，找到插入的位置即可，當(dāng)然方法不止這一個(gè)，也可以是通知的時(shí)候選取優(yōu)先級最大的通知。我覺得基于此我們可以把juc的所有類都可以實(shí)現(xiàn)。
上面都是基于wait/notify/notifyAll來同步的。wait/notify機(jī)制有個(gè)很蛋疼的地方是，比如線程B要用notify通知線程A，那么線程B要確保線程A已經(jīng)在wait調(diào)用上等待了，否則線程A可能永遠(yuǎn)都在等待。編程的時(shí)候就會很蛋疼。另外，是調(diào)用notify，還是notifyAll？notify只會喚醒一個(gè)線程，如果錯(cuò)誤地有兩個(gè)線程在同一個(gè)對象上wait等待，那么又悲劇了。為了安全起見，貌似只能調(diào)用notifyAll了
看一看 java.util.concurrent.locks對wait/notify/notifyAll的代替 怎么實(shí)現(xiàn)的各種鎖

Paste_Image.png

這里涉及到一個(gè)基礎(chǔ)類 也是基于Unsafe 類實(shí)現(xiàn)的。
給出官方api的翻譯版

用來創(chuàng)建鎖和其他同步類的基本線程阻塞原語。
此類以及每個(gè)使用它的線程與一個(gè)許可關(guān)聯(lián)（從 Semaphore 類的意義上說）。如果該許可可用，并且可在進(jìn)程中使用，則調(diào)用 park 將立即返回；否則可能 阻塞。如果許可尚不可用，則可以調(diào)用 unpark 使其可用。（但與 Semaphore 不同的是，許可不能累積，并且最多只能有一個(gè)許可。）
park 和 unpark 方法提供了阻塞和解除阻塞線程的有效方法，并且不會遇到導(dǎo)致過時(shí)方法 Thread.suspend 和 Thread.resume 因?yàn)橐韵履康淖兊貌豢捎玫膯栴}：由于許可的存在，調(diào)用 park 的線程和另一個(gè)試圖將其 unpark 的線程之間的競爭將保持活性。此外，如果調(diào)用者線程被中斷，并且支持超時(shí)，則 park 將返回。park 方法還可以在其他任何時(shí)間“毫無理由”地返回，因此通常必須在重新檢查返回條件的循環(huán)里調(diào)用此方法。從這個(gè)意義上說，park 是“忙碌等待”的一種優(yōu)化，它不會浪費(fèi)這么多的時(shí)間進(jìn)行自旋，但是必須將它與 unpark 配對使用才更高效。
三種形式的 park 還各自支持一個(gè) blocker 對象參數(shù)。此對象在線程受阻塞時(shí)被記錄，以允許監(jiān)視工具和診斷工具確定線程受阻塞的原因。（這樣的工具可以使用方法 getBlocker(java.lang.Thread) 訪問 blocker。）建議最好使用這些形式，而不是不帶此參數(shù)的原始形式。在鎖實(shí)現(xiàn)中提供的作為 blocker 的普通參數(shù)是 this。
這些方法被設(shè)計(jì)用來作為創(chuàng)建高級同步實(shí)用工具的工具，對于大多數(shù)并發(fā)控制應(yīng)用程序而言，它們本身并不是很有用。park 方法僅設(shè)計(jì)用于以下形式的構(gòu)造：
while (!canProceed()) { ... LockSupport.park(this); }在這里，在調(diào)用 park 之前，canProceed 和其他任何動作都不會鎖定或阻塞。因?yàn)槊總€(gè)線程只與一個(gè)許可關(guān)聯(lián)，park 的任何中間使用都可能干擾其預(yù)期效果。
示例用法。 以下是一個(gè)先進(jìn)先出 (first-in-first-out) 非重入鎖類的框架。

class FIFOMutex {
   private final AtomicBoolean locked = new AtomicBoolean(false);
   private final Queue<Thread> waiters
     = new ConcurrentLinkedQueue<Thread>();
   public void lock() {
     boolean wasInterrupted = false;
     Thread current = Thread.currentThread();
     waiters.add(current);
    // Block while not first in queue or cannot acquire lock
     while (waiters.peek() != current ||
            !locked.compareAndSet(false, true)) {
        LockSupport.park(this);
        if (Thread.interrupted()) // ignore interrupts while waiting
          wasInterrupted = true;
     }
     waiters.remove();
     if (wasInterrupted)          // reassert interrupt status on exit
        current.interrupt();
   }
   public void unlock() {
     locked.set(false);
     LockSupport.unpark(waiters.peek());
   }
 }

這里寫了個(gè)測試類，附上源碼(有大概注釋)

package com.alibaba.otter.canal.common;

import java.util.concurrent.locks.LockSupport;

import org.junit.After;
import org.junit.Before;
import org.junit.Test;
public class LockSupportTest extends AbstractZkTest {
    @Before
    public void setUp() {
        
    }

    @After
    public void tearDown() {
    }

    @Test
    public void testLockSupport() {
         LockSupport.park();
         System.out.println("block.");//阻塞在這里證明 默認(rèn)許可時(shí)不可用的
    }
    @Test
    public void testUnpark() {
         Thread thread = Thread.currentThread();
         LockSupport.unpark(thread);//釋放許可
         LockSupport.park();// 獲取許可
         System.out.println("b");//正常執(zhí)行 一對一使用
    }
    
    @Test
    public void testReentrantUnpark() {
        Thread thread = Thread.currentThread();
        
        LockSupport.unpark(thread);
        
        System.out.println("a");
        LockSupport.park();
        System.out.println("b");
        LockSupport.park();
        System.out.println("c");//阻塞在這里 ，說明非可重入的
    }
    @Test
    public void testInterrupt() throws Exception {
        Thread t = new Thread(new Runnable()
        {
            private int count = 0;

            @Override
            public void run()
            {
                long start = System.currentTimeMillis();
                long end = 0;

                while ((end - start) <= 1000)
                {
                    count++;
                    end = System.currentTimeMillis();
                }

                System.out.println("after 1 second.count=" + count);

            //等待或許許可
                LockSupport.park();
                System.out.println("thread over." + Thread.currentThread().isInterrupted());

            }
        });

        t.start();

        Thread.sleep(2000);

        // 中斷線程
        t.interrupt(); //不會拋出InterruptException 不影響主線程

        
        System.out.println("main over");
    }
    
    
    
}
最后編輯于 ：
?著作權(quán)歸作者所有,轉(zhuǎn)載或內(nèi)容合作請聯(lián)系作者
【社區(qū)內(nèi)容提示】社區(qū)部分內(nèi)容疑似由AI輔助生成，瀏覽時(shí)請結(jié)合常識與多方信息審慎甄別。
平臺聲明：文章內(nèi)容（如有圖片或視頻亦包括在內(nèi)）由作者上傳并發(fā)布，文章內(nèi)容僅代表作者本人觀點(diǎn)，簡書系信息發(fā)布平臺，僅提供信息存儲服務(wù)。

相關(guān)閱讀更多精彩內(nèi)容

  • 1.解決信號量丟失和假喚醒 public class MyWaitNotify3{ MonitorObject m...
    Q羅閱讀 993評論 0贊 1
  • 本文出自 Eddy Wiki ，轉(zhuǎn)載請注明出處：http://eddy.wiki/interview-java.h...
    eddy_wiki閱讀 2,297評論 0贊 14
  • layout: posttitle: 《Java并發(fā)編程的藝術(shù)》筆記categories: Javaexcerpt...
    xiaogmail閱讀 6,005評論 1贊 19
  • 人是會懷念的，并非要到特定的老齡或者身體機(jī)能漸漸退化了才細(xì)想過去，而后被動拉扯些舊日里的光景潤色眼下局限著...
    及客閱讀 240評論 0贊 2
  • 早秋的天空顏色總是很淡，穹頂?shù)挠羲{(lán)淺薄的近乎透明；云彩像斑斑魚鱗又像裸露的河床，分來一點(diǎn)夕陽的秋意，散發(fā)著干凈松爽...
    小野風(fēng)信子閱讀 494評論 0贊 2

友情鏈接更多精彩內(nèi)容