mysql的鎖--行鎖，表鎖，樂觀鎖，悲觀鎖

時間 2019-11-25

標籤 mysql 樂觀悲觀欄目 MySQL 简体版

原文原文鏈接

一引言--爲何mysql提供了鎖html

　　最近看到了mysql有行鎖和表鎖兩個概念，越想越疑惑。爲何mysql要提供鎖機制，並且這種機制不是一個擺設，還有不少人在用。在現代數據庫裏幾乎有事務機制，acid的機制應該能解決併發調度的問題了，爲何還要主動加鎖呢？mysql

　　後來看到一篇文章，「防止更新丟失，並不能單靠數據庫事務控制器來解決，須要應用程序對要更新的數據加必要的鎖來解決」。瞬間，世界觀都崩塌了。很是不敢相信，因而本身寫了代碼檢驗一下。sql

　　數據庫表是這樣的。用count字段來作100次累加。數據庫

　　爲了保證明驗的科學性，先確認了數據庫是InnoDB的，這樣纔有事務機制；也確認了隔離性級別編程

定義一個任務，讀count值--程序count++--寫數據庫安全

public class LostUpdate implements Runnable{
    private CountDownLatch countDown;
    public LostUpdate(CountDownLatch countDown){
        this.countDown = countDown;
    }
    
    @Override
    public void run() {
        Connection conn=null;
        try {
            Class.forName("com.mysql.jdbc.Driver");
            conn = DriverManager.getConnection("jdbc:mysql://localhost:3306/test?useUnicode=true&characterEncoding=UTF-8",
                    "root", "123");
        } catch (Exception e) {
            e.printStackTrace();
            return;
        }
        
        try {
            conn.setAutoCommit(false);
            //不加鎖的狀況
            PreparedStatement ps =conn.prepareStatement("select * from LostUpdate where id =1");
            //加鎖的狀況
            //PreparedStatement ps =conn.prepareStatement("select * from LostUpdate where id =1 for update");
            ResultSet rs=ps.executeQuery();
            int count = 0;
            while(rs.next()){
                count= rs.getInt("count");
            }
            
            count++;
            ps =conn.prepareStatement("update LostUpdate set count=? where id =1");
            ps.setInt(1, count);
            ps.executeUpdate();
            
            conn.commit();
        } catch (Exception e) {
            try {
                conn.rollback();
            } catch (SQLException e1) {
                e1.printStackTrace();
            }
            e.printStackTrace();
        }
        //表示一次任務完成
        countDown.countDown();
    }
}

主線程下建立子線程，模擬多線程環境數據結構

public class TestLock {    
    public static void main(String[] args) throws InterruptedException {
        //建立線程池，裏面有10個線程，共執行100次+1操做
        final int THREAD_COUNT=10;
        final int RUN_TIME=100;
        
        ExecutorService threadPool=Executors.newFixedThreadPool(THREAD_COUNT);
        //用CountDownLatch保證主線程等待全部任務完成
        CountDownLatch count=new CountDownLatch(RUN_TIME);
        
        for(int i=0;i<RUN_TIME;i++)
            threadPool.execute(new LostUpdate(count));
        
        threadPool.shutdown();
        count.await();
        //提示全部任務執行完
        System.out.println("finish");
    }
}

運行結果是：多線程

　　大概解釋一下程序，就是建立了一個線程池，裏面10個線程，執行100次任務。每一個任務就是讀count值--程序count++--寫數據庫，經典的銀行存款（丟失修改）問題。事實勝於雄辯，結論就是上面的橙色字，解決丟失修改不能靠事務，要加必要的鎖，因此數據庫提供的鎖不是個擺設。併發

二數據庫事務機制mvc

　　爲了找到問題的根源，爲了拯救我崩潰的世界觀，我又去回顧了數據庫事務的知識。借鑑這篇

　　數據庫的acid屬性

原性性（Actomicity）：事務是一個原子操做單元，其對數據的修改，要麼全都執行，要麼全都不執行。
一致性（Consistent）：在事務開始和完成時，數據都必須保持一致狀態。這意味着全部相關的數據規則都必須應用於事務的修改，以操持完整性；事務結束時，全部的內部數據結構（如B樹索引或雙向鏈表）也都必須是正確的。
隔離性（Isolation）：數據庫系統提供必定的隔離機制，保證事務在不受外部併發操做影響的「獨立」環境執行。這意味着事務處理過程當中的中間狀態對外部是不可見的，反之亦然。
持久性（Durable）：事務完成以後，它對於數據的修改是永久性的，即便出現系統故障也可以保持。

　　說好的一致性呢，童話裏都是騙人的！！　　

　　事務併發調度的問題

髒讀（dirty read）：A事務讀取B事務還沒有提交的更改數據，並在這個數據基礎上操做。若是B事務回滾，那麼A事務讀到的數據根本不是合法的，稱爲髒讀。在oracle中，因爲有version控制，不會出現髒讀。
不可重複讀（unrepeatable read）：A事務讀取了B事務已經提交的更改（或刪除）數據。好比A事務第一次讀取數據，而後B事務更改該數據並提交，A事務再次讀取數據，兩次讀取的數據不同。
幻讀（phantom read）：A事務讀取了B事務已經提交的新增數據。注意和不可重複讀的區別，這裏是新增，不可重複讀是更改（或刪除）。這兩種狀況對策是不同的，對於不可重複讀，只須要採起行級鎖防止該記錄數據被更改或刪除，然而對於幻讀必須加表級鎖，防止在這個表中新增一條數據。
第一類丟失更新：A事務撤銷時，把已提交的B事務的數據覆蓋掉。
第二類丟失更新：A事務提交時，把已提交的B事務的數據覆蓋掉。

　　三級封鎖協議

一級封鎖協議：事務T中若是對數據R有寫操做，必須在這個事務中對R的第一次讀操做前對它加X鎖，直到事務結束才釋放。事務結束包括正常結束（COMMIT）和非正常結束（ROLLBACK）。
二級封鎖協議：一級封鎖協議加上事務T在讀取數據R以前必須先對其加S鎖，讀完後方可釋放S鎖。
三級封鎖協議：一級封鎖協議加上事務T在讀取數據R以前必須先對其加S鎖，直到事務結束才釋放。

　　可見，三級鎖操做一個比一個厲害（知足高級鎖則必定知足低級鎖）。但有個很是致命的地方，一級鎖協議就要在第一次讀加x鎖，直到事務結束。幾乎就要在整個事務加寫鎖了，效率很是低。三級封鎖協議只是一個理論上的東西，實際數據庫經常使用另外一套方法來解決事務併發問題。

　　隔離性級別

　　mysql用意向鎖（另外一種機制）來解決事務併發問題，爲了區別封鎖協議，弄了一個新概念隔離性級別：包括Read Uncommitted、Read Committed、Repeatable Read、Serializable，見這篇。mysql 通常默認Repeatable Read。

　　終於發現本身爲何會誤會事務能解決丟失修改了。至於爲何隔離性級別不解決丟失修改，我猜是有更好的解決方案吧。

　　總結一下，repeatable read能解決髒讀和不可重複讀，但不嗯呢該解決丟失修改。

三 mysql的行鎖和表鎖

　　說了那麼久，終於入正題了，先來講說什麼是行鎖和表鎖。

表級鎖：每次操做鎖住整張表。開銷小，加鎖快；不會出現死鎖；鎖定粒度大，發生鎖衝突的機率最高，併發度最低；
行級鎖：每次操做鎖住一行數據。開銷大，加鎖慢；會出現死鎖；鎖定粒度最小，發生鎖衝突的機率最低，併發度也最高；
頁面鎖：開銷和加鎖時間界於表鎖和行鎖之間；會出現死鎖；鎖定粒度界於表鎖和行鎖之間，併發度通常。沒弄懂，有空再看。？

1 MyISAM的鎖

　　稍微提一下MyISAM，只說和InnoDB不一樣的。

　　a. MyISAM只有表鎖，鎖又分爲讀鎖和寫鎖。　

　　b. 沒有事務，不用考慮併發問題，世界和平~

　　c. 因爲鎖的粒度太大，因此當該表寫併發量較高時，要等待的查詢就會不少了。優化見這裏。

2 InnoDB的行鎖和表鎖

　　沒有特定的語法。mysql的行鎖是經過索引體現的，參考。

　　若是where條件中只用到索引項，則加的是行鎖；不然加的是表鎖。好比說主鍵索引，惟一索引和聚簇索引等。若是sql的where是全表掃描的，想加行鎖也心有餘而力不足。

　　行鎖和表鎖對咱們編程有什麼影響，要在where中儘可能只用索引項，不然就會觸發表鎖。另外一個多是，咱們發瘋了地想優化查詢，但where子句中就是有非索引項，因而咱們本身寫鏈接？

　　行鎖和表鎖各適合怎麼樣的應用，待求證？。

3 讀鎖和寫鎖

　　InnoDB用意向鎖？實現隔離性級別，原理未名，貼張圖：

　　回想鎖協議，對什麼操做加什麼鎖是一個問題，加鎖加到何時有是一個問題。鎖協議裏經常會看到「加鎖直到事務結束」的煩心字樣。而在InnoDB中，select,insert,update,delete等語句執行時都會自動加解鎖。select的鎖通常執行完就釋放了，修改操做的X鎖會持有到事務結束，效率高不少。至於詳細的加鎖原理，見這裏，搜「InnoDB存儲引擎中不一樣SQL在不一樣隔離級別下鎖比較」

　　mysql也給用戶提供了加鎖的機會，只要在sql後加LOCK IN SHARE MODE 或FOR UPDATE

　　共享鎖（S）：SELECT * FROM table_name WHERE ... LOCK IN SHARE MODE
　　排他鎖（X）：SELECT * FROM table_name WHERE ... FOR UPDATE

　　值得注意的是，本身加的鎖沒有釋放鎖的語句，因此鎖會持有到事務結束。

　　mysql 還提供了LOCK TABLES，UNLOCK TABLES，用於加表鎖，怎麼用還不太清楚？

4 考察加鎖的狀況

　　加了讀鎖仍是寫鎖，加了行鎖仍是表鎖，說何時釋放，能夠從原理上分析。但剛開始時我不太懂原理，因而又寫了個程序。

public class ForUpdate1  implements Runnable{
    private CountDownLatch countDown;
    public ForUpdate1(CountDownLatch countDown){
        this.countDown = countDown;
    }
    @Override
    public void run() {
        Connection conn=null;
        try {
            Class.forName("com.mysql.jdbc.Driver");
            conn = DriverManager.getConnection("jdbc:mysql://localhost:3306/test?useUnicode=true&characterEncoding=UTF-8",
                    "root", "123");
        } catch (Exception e) {
            e.printStackTrace();
            return;
        }
        
        try {
            conn.setAutoCommit(false);
            /*PreparedStatement ps =conn.prepareStatement("select * from LostUpdate where id =1 for update");
            ps.executeQuery();*/        
            PreparedStatement ps =conn.prepareStatement("update LostUpdate set count =1 where id =1");
            ps.executeUpdate();
            Thread.sleep(10000);
            
            conn.commit();
            System.out.println("test 1 finish");
            countDown.countDown();
        } catch (Exception e) {
            try {
                conn.rollback();
            } catch (SQLException e1) {
                e1.printStackTrace();
            }
            e.printStackTrace();
        }    
    }
}

public class ForUpdate2  implements Runnable{
    private CountDownLatch countDown;
    public ForUpdate2(CountDownLatch countDown){
        this.countDown = countDown;
    }
    
    @Override
    public void run() {
        Connection conn=null;
        try {
            Class.forName("com.mysql.jdbc.Driver");
            conn = DriverManager.getConnection("jdbc:mysql://localhost:3306/test?useUnicode=true&characterEncoding=UTF-8",
                    "root", "123");
        } catch (Exception e) {
            e.printStackTrace();
            return;
        }
        
        try {
            Thread.sleep(2000);
            conn.setAutoCommit(false);
            PreparedStatement ps =conn.prepareStatement("select * from LostUpdate where id =1 for update");
            ps.executeQuery();
            /*PreparedStatement ps =conn.prepareStatement("update LostUpdate set count =1 where id =1");
            ps.executeUpdate();*/        
            
            conn.commit();
            System.out.println("test 2 finish");
            countDown.countDown();
        } catch (Exception e) {
            try {
                conn.rollback();
            } catch (SQLException e1) {
                e1.printStackTrace();
            }
            e.printStackTrace();
        }    
    }
}

public class TestForUpdate {    
    public static void main(String[] args) throws InterruptedException {
        final int THREAD_COUNT=10;
        
        ExecutorService threadPool=Executors.newFixedThreadPool(THREAD_COUNT);
        CountDownLatch count=new CountDownLatch(2);
        
        threadPool.execute(new ForUpdate1(count));
        threadPool.execute(new ForUpdate2(count));
        
        threadPool.shutdown();
        count.await();
        System.out.println("finish");
    }
}

　　只有兩個線程，ForUpdate1先執行sql語句以後等10s，ForUpdate2先等待2s再執行sql語句。因此若是ForUpdate1持有鎖，並且ForUpdate2等待，輸出就應該是test 1 finish->test 2 finish->finish；不然就是test 2 finish->test 1 finish->finish。

　　這個程序改一下能測試上面說的理論：

repeatable read能解決髒讀和不可重複讀
好比行鎖真的只鎖住一行
s，x，is和ix的關係

　　判斷加鎖狀況，mysql應該有工具，但沒找到？

　　能夠經過檢查InnoDB_row_lock狀態變量來分析系統上的行鎖的爭奪狀況：
　　mysql> show status like 'innodb_row_lock%';

　　若是發現鎖爭用比較嚴重，如InnoDB_row_lock_waits和InnoDB_row_lock_time_avg的值比較高，還能夠經過設置InnoDB Monitors來進一步觀察發生鎖衝突的表、數據行等，並分析鎖爭用的緣由。不明覺厲？，看這篇

　　總結一下這一章，mysql提供了行鎖和表鎖，咱們寫語句時應該儘可能啓動行鎖，以提升效率；另外一方面，也說了一下讀鎖和寫鎖的原理。好了武器（原理）咱們都懂了，那就看怎麼優化了。

四解決丟失修改--樂觀鎖和悲觀鎖

　　首先爲何要加鎖，加鎖就是爲了解決丟失修改（也不知道這麼說對不對）。若是一個事務中只有一句sql，數據庫是能夠保證它是併發安全的。丟失修改的特徵就是在一個事務中先讀P數據，再寫P數據，注意是同一個數據（也不知道這麼說對不對）。只是本身推理了一下，沒有太強的理據。所謂丟失修改，通常是A事務有兩個操做，後一個操做依賴於前一個操做，以後後一個操做覆蓋了B事務的寫操做，能夠表示爲這樣。

　　pro1多是Read(P),Write(P),Read(Q),Write(Q),其中P=2Q，數據庫中的冗餘致使的關聯關係是很常見的。

pro1=Read(P)，就是咱們結論中的狀況；
pro1=Write(P)，pro1處會對P加IX鎖？，IX鎖會直至事務結束，不會丟失修改；
pro1=Read(Q)或Write(Q)，雖然語法上回發生這種狀況，但既然數據時關聯的，那在兩個事務中都應該同時操做P，Q。這樣就規範到第一種狀況。

　　綜上，若是一個事務先讀後寫同一份數據，就可能發生丟失修改，要作一些處理。能夠用下面的樂觀鎖和悲觀鎖解決。

　　悲觀鎖和樂觀鎖的概念：

　　悲觀鎖（Pessimistic Concurrency Control，PCC）：假定會發生併發衝突，屏蔽一切可能違反數據完整性的操做。至於怎麼加鎖，加鎖的範圍也沒講。

　　樂觀鎖（Optimistic Concurrency Control，OCC）：假設不會發生併發衝突，只在提交操做時檢查是否違反數據完整性。也沒具體指定怎麼檢查。

　　就是這麼概念，什麼都不說清楚。畢竟樂觀鎖和悲觀鎖也不只僅能用在數據庫中，也能用在線程中。

　　悲觀的缺陷是不管是頁鎖仍是行鎖，加鎖的時間可能會很長，這樣可能會長時間的限制其餘用戶的訪問，也就是說悲觀鎖的併發訪問性很差。

　　樂觀鎖不能解決髒讀，加鎖的時間要比悲觀鎖短（只是在執行sql時加了基本的鎖保證隔離性級別），樂觀鎖能夠用較大的鎖粒度得到較好的併發訪問性能。可是若是第二個用戶剛好在第一個用戶提交更改以前讀取了該對象，那麼當他完成了本身的更改進行提交時，數據庫就會發現該對象已經變化了，這樣，第二個用戶不得不從新讀取該對象並做出更改。

　　可見，樂觀鎖更適合解決衝突機率極小的狀況；而悲觀鎖則適合解決併發競爭激烈的狀況，儘可能用行鎖，縮小加鎖粒度，以提升併發處理能力，即使加行鎖的時間比加表鎖的要長。

　　悲觀鎖的例子

　　並無人說悲觀鎖要怎麼加鎖，加鎖的範圍如何。這裏僅僅提供一種解決丟失修改的悲觀鎖例子。

　　丟失修改咱們用第一章講到的累積100次的例子。綜合前面講到的結論，丟失修改的特徵就是在一個事務中先讀P數據，再寫P數據。並且一級鎖協議能解決丟失修改，因此若是事務A 中寫P，咱們只要在A中第一次讀P前加X鎖。作法在第一章程序中有：

//把
PreparedStatement ps =conn.prepareStatement("select * from LostUpdate where id =1");
//換成
PreparedStatement ps =conn.prepareStatement("select * from LostUpdate where id =1 for update");

　　樂觀鎖的例子

　　樂觀鎖也沒有指定怎麼檢測併發衝突，下面是常見的兩種作法（參考）：

使用數據版本（Version）。在P數據上（一般每一行）加version字段(int)，A事務在讀數據P 時同時讀出版本號，在修改數據前檢測最新版本號是否等於先前取出的版本號，若是是，則修改，同時把版本號+1；不然要麼回滾，要麼從新執行事務。另外，數據P的全部修改操做都要把版本號+1。有一個很是重要的點，版本號是用來查看被讀的變量有無變化，而不是針對被寫的變量，做用是防止被依賴的變量有修改。
使用時間戳(TimeStamp)。作法相似於1中。

　　下面寫兩個例子，背景仍是一開始的累積100次的丟失修改問題，都是用version解決的。

　　1 當發生衝突時回滾並拋異常

　　任務類

public class LostUpdateOccDiscard implements Runnable{
    private CountDownLatch countDown;
    public LostUpdateOccDiscard(CountDownLatch countDown){
        this.countDown = countDown;
    }
    
    @Override
    public void run() {
        Connection conn=null;
        try {
            Class.forName("com.mysql.jdbc.Driver");
            conn = DriverManager.getConnection("jdbc:mysql://localhost:3306/test?useUnicode=true&characterEncoding=UTF-8",
                    "root", "123");
        } catch (Exception e) {
            e.printStackTrace();
            return;
        }
        
        try {
            conn.setAutoCommit(false);
            //讀的時候一併讀出version
            PreparedStatement ps =conn.prepareStatement("select * from LostUpdate where id =1");
            ResultSet rs=ps.executeQuery();
            int count = 0;
            int version = 0;
            while(rs.next()){
                count= rs.getInt("count");
                version= rs.getInt("version");
            }
            
            count++;
            
            //更新操做，用cas原子操做來更新
            ps =conn.prepareStatement("update LostUpdate set count=?, version=version+1 where id =1 and version=?");
            ps.setInt(1, count);
            ps.setInt(2, version);
            int result = ps.executeUpdate();
            
            //檢查有無因衝突致使執行失敗
            //成功，則commit，完成任務
            if(result>0) {    
                conn.commit();
            }
            //失敗，回滾，拋異常提醒調用者出現衝突。
            else{
                conn.rollback();
                throw new Exception("更新count出現衝突");
            }            
        } catch (SQLException e) {
            try {
                conn.rollback();
            } catch (SQLException e1) {
                e1.printStackTrace();
            }
            e.printStackTrace();
        }
        catch (Exception e) {
            System.out.println(e.getMessage());
        }
        //表示一次任務完成
        countDown.countDown();
    }
}

　　　主線程，和前面差很少，建立10個線程，執行100個任務。

public class TestLockOcc {    
    public static void main(String[] args) throws InterruptedException {
        //建立線程池，裏面有10個線程，共執行100次+1操做
        final int THREAD_COUNT=10;
        final int RUN_TIME=100;
        
        ExecutorService threadPool=Executors.newFixedThreadPool(THREAD_COUNT);
        //用CountDownLatch保證主線程等待全部任務完成
        CountDownLatch count=new CountDownLatch(RUN_TIME);
        
        for(int i=0;i<RUN_TIME;i++)
            threadPool.execute(new LostUpdateOccDiscard(count));
        
        threadPool.shutdown();
        count.await();
        //提示全部任務執行完
        System.out.println("finish");
    }
}

View Code

　　輸出結果：在console裏出了一堆異常，看數據庫，大概累積了10-12次

　　不要懷疑，程序沒有問題。

　　a. 對着上面說的version方法的原理，程序也比較好懂。

　　b. 更新時要用cas（compare and set）的原子操做，一步搞定。而不是先讀一次version，比較完再執行依據update。想一想也知道後者在多線程有問題。

　　至於爲何只累積了10-12次，緣由是這個累加的併發量是10，就是有10個線程在爭奪着修改權。九死一輩子啊，1個線程commit了，就意味着9個線程要rollback拋異常。

　　2 當發生衝突時重試，有時咱們咱們不但願程序裏那麼多異常

　　任務類　　

public class LostUpdateOcc implements Runnable{
    private CountDownLatch countDown;
    public LostUpdateOcc(CountDownLatch countDown){
        this.countDown = countDown;
    }
    
    @Override
    public void run() {
        Connection conn=null;
        try {
            Class.forName("com.mysql.jdbc.Driver");
            conn = DriverManager.getConnection("jdbc:mysql://localhost:3306/test?useUnicode=true&characterEncoding=UTF-8",
                    "root", "123");
        } catch (Exception e) {
            e.printStackTrace();
            return;
        }
        
        try {            
            int try_times=100;
            int count;
            int version;    
            PreparedStatement ps;
            ResultSet rs;
            
            //把循環條件放在裏面if裏
            while(try_times>0){
                //開始事務
                try_times--;
                conn.setAutoCommit(false);
                
                //讀操做
                ps=conn.prepareStatement("select * from LostUpdate where id =1");
                rs=ps.executeQuery();
                
                //判斷事務執行的條件，首先是能執行，其次是須要執行
                if(rs.next()){
                    count= rs.getInt("count");
                    version= rs.getInt("version");
                    
                    count++;
                    
                    //更新操做，用cas原子操做來更新
                    ps =conn.prepareStatement("update LostUpdate set count=?, version=version+1 where id =1 and version=?");
                    ps.setInt(1, count);
                    ps.setInt(2, version);
                    int result = ps.executeUpdate();
                    
                    //每次執行完更新操做，檢測一次衝突
                    //成功，則繼續事務
                    //失敗，回滾，睡100ms，避開競爭。結束此次循環，開啓新事務。
                    if(result==0) {    
                        conn.rollback();
                        Thread.sleep(100);
                        continue;
                    }
                    
                    //事務一帆風順，沒遇到衝突，事務提交，跳出while
                    conn.commit();
                    break;
                }
                //做爲while條件不成立時的處理，好比該行數據被刪除。
                else{
                    conn.rollback();
                    break;
                }                                                            
            }
            if(try_times<=0) throw new Exception("衝突重試的此時過多，事務失敗");
            System.out.println(try_times);
        } catch (SQLException e) {
            try {
                conn.rollback();
            } catch (SQLException e1) {
                e1.printStackTrace();
            }
            e.printStackTrace();
        }catch (Exception e) {
            System.out.println(e.getMessage());
        }
        
        //表示一次任務完成
        countDown.countDown();
    }
}

　　主線程，和前面差很少，建立10個線程，執行100個任務。

public class TestLockOcc {    
    public static void main(String[] args) throws InterruptedException {
        //建立線程池，裏面有10個線程，共執行100次+1操做
        final int THREAD_COUNT=10;
        final int RUN_TIME=100;
        
        ExecutorService threadPool=Executors.newFixedThreadPool(THREAD_COUNT);
        //用CountDownLatch保證主線程等待全部任務完成
        CountDownLatch count=new CountDownLatch(RUN_TIME);
        
        for(int i=0;i<RUN_TIME;i++)
            threadPool.execute(new LostUpdateOcc(count));
        
        threadPool.shutdown();
        count.await();
        //提示全部任務執行完
        System.out.println("finish");
    }
}