MySQL實現分散式鎖

2022-08-01 14:04:14

基於MySQL分散式鎖實現原理及程式碼

工欲善其事必先利其器，在基於MySQL實現分散式鎖之前，我們要先了解一點MySQL鎖自身的相關內容

MySQL鎖

我們知道：鎖是計算機協調多個程序或者執行緒並行存取同一資源的機制，而在資料庫中，除了傳統的機器資源的爭用之外，儲存下來的資料也屬於供使用者共用的資源，所以如何保證資料並行的一致性，有效性是每個資料庫必須解決的問題。

除此之外，鎖衝突也是影響資料庫並行效能的主要因素，所以鎖對於資料庫而言就顯得非常重要，也非常複雜。

而儲存引擎是MySQL中非常重要的底層元件，主要用來處理不同型別的SQL操作，其中包括建立，讀取，刪除和修改操作。在MySQL中提供了不同型別的儲存引擎，根據其不同的特性提供了不同的儲存機制，索引和鎖功能。

根據show engines;能夠列出MySQL下支援的儲存引擎

如果沒有特殊指定，那麼在MySQL8.0中會設定InnoDB為預設的儲存引擎

在實際工作中，根據需求選擇最多的兩種儲存引擎分別為：

InnoDB
MyISAM

所以我們主要針對這兩種型別來介紹MySQL的鎖

InnoDB

InnoDB支援多粒度鎖定，可以支援行鎖，也可以支援表鎖。如果沒有升級鎖粒度，那麼預設情況下是以行鎖來設計的。

關於行鎖和表鎖的介紹：

行鎖對指定資料進行加鎖，鎖定粒度最小，開銷大，加鎖慢，容易出現死鎖問題，出現鎖衝突的概率最小，並行性最高
表鎖對整個表進行加鎖，鎖定粒度大，開銷小，加鎖快，不會出現死鎖，出現鎖衝突的概率最大，並行性最低

這裡沒法說明那種鎖最好，只有合適不合適

在行級鎖中，可以分為兩種型別

共用鎖
排他鎖

共用鎖

共用鎖又稱為讀鎖，允許其他事務讀取被鎖定的物件，也可以在其上獲取其他共用鎖，但不能寫入。

舉個例子：

事務T在資料A擁有共用鎖，那麼當前事務T對資料A可以讀，但是不能修改。而且事務T2同樣可以對資料A擁有共用鎖，這樣相當於在資料A上分別存在不同事務的共用鎖
資料A擁有了事務T的共用鎖，那麼就不能再擁有其他事務的排他鎖

下面是關於共用鎖的具體實現，關鍵程式碼:select .. from table lock in share mode

 -- 建立範例表
 create table tb_lock(
     id bigint primary key auto_increment,
     t_name varchar(20)
 ) engine=InnoDB;

開啟兩個視窗來測試：

session1	session2
set autocommit=0;	set autocommit=0;
select * from tb_lock where t_name = ‘zs’ lock in share mode;
	select * from tb_lock where t_name = ‘zs’ lock in share mode;
	select * from tb_lock where t_name = ‘lsp’ lock in share mode;
update tb_lock set t_name = ‘lzs’ where t_name = ‘zs’;
update tb_lock set t_name = ‘lsp111’ where t_name = ‘lsp’;
	select * from tb_lock where t_name = ‘zs’;
commit;

自動提交全部關閉，可以通過select @@autocommit;來檢視

通過以上實驗，我們總結:

共用鎖基於行鎖處理，不同事務可以在同一條資料上獲取共用鎖
如果多個事務在同一條資料上獲取共用鎖，當想要修改該條資料的時候，會出現阻塞狀態。直到其他事務將鎖釋放，該能夠繼續修改

修改，刪除，插入會預設對涉及到的資料加上排他鎖

單純的select操作不會有任何影響，select不會加任何鎖
執行commit;自動釋放鎖

排它鎖

又叫寫鎖。只允許獲取鎖的事務對資料進行操作【更新，刪除】，其他事務對相同資料集只能進行讀取，不能有跟新或者刪除操作。而且也不能在相同資料集獲取到共用鎖。

沒錯，就是這麼霸道

在MySQL中，想要基於排它鎖實現行級鎖，就需要對錶中索引列加鎖，否則的話，排它鎖就屬於表級鎖

下面一一來展示，關鍵程式碼：select .. from XX for update

首先是有索引列狀態

session1	session2
set autocommit=0;	set autocommit=0;
select * from tb_lock;	select * from tb_lock;
select * from tb_lock where id = 1 for update;
	select * from tb_lock where id = 1 for update;
select * from tb_lock where id = 2 for update;
commit;

通過以上實驗，得到結論：

對索引列進行加鎖的鎖定級別為行級鎖，如上所示，當其他事務想要對相同的資料再次加鎖的時候，就會進行到阻塞狀態。並且如果等待時間過長，會出現如下異常：

 Lock wait timeout exceeded; try restarting transaction

對不同行資料再次加排它鎖，是沒有任何問題的。
對已經上鎖的相同資料做修改和刪除操作不需要多說，因為InnoDB預設會對其加入排它鎖

下面是無索引列狀態

session1	session2
set autocommit=0;	set autocommit=0;
select * from tb_lock;	select * from tb_lock;
select * from tb_lock where t_name = ‘ls’ for update;
	select * from tb_lock where t_name = ‘ls’ for update;
commit

通過以上實驗，得到結論：

對非索引列其中一條資料加入了排它鎖後，在其他事務中對不同資料再次加入排它鎖，進入了阻塞狀態
說明當加鎖列屬於非索引時，InnoDB會對整個表進行上鎖，進入到表級鎖

接下來我們來看看MyISAM的方式

MyISAM

MyISAM屬於表級鎖，被用來防止任何其他事務存取表的鎖。

其中表鎖又分為兩種形式

表共用讀鎖： READ
表獨佔寫鎖： WRITE

這裡我們要注意：表級鎖只能防止其他對談進行不適當的讀取或寫入。

持有WRITE 鎖的對談可以執行表級操作，比如DELETE或者TRUNCATE
持有對談READ鎖，不能夠執行DELETE或者TRUNCATE操作

表共用讀鎖

不管是READ還是WRITE，都是通過lock table 來獲取表鎖的，而READ鎖擁有如下特性：

持有鎖的對談可以讀取表，但是不能進行寫入操作
多個對談可以同時獲取READ表的鎖，而其他對談可以在不顯式獲取READ鎖的情況下讀取該表：也就是說直接通過select來操作

那麼，接下來我們來看實際操作，關鍵程式碼：lock tables table_name read

 create table tb_lock_isam(
     id bigint primary key auto_increment,
     t_name varchar(20)
 ) engine=MyISAM;

開啟兩個視窗來進行操作：

session1	session2
set autocommit=0;	set autocommit=0;
LOCK TABLES tb_lock_isam READ;
select * from tb_lock_isam;
select * from tb_lock;
	select * from tb_lock_isam;
	LOCK TABLES tb_lock_isam READ;
	select * from tb_lock_isam;
	select * from tb_lock;
unlock tables;	insert into tb_lock_isam(t_name) values(‘ll’);

通過以上實戰，驗證以下結論：

在當前事務下，獲取到讀鎖，直接查詢鎖定表是沒有問題的，但是如果想要讀取其他表下的資料，那麼就會出現以下異常：因為其他表並沒有LOCK在其中

 Table 'tb_lock' was not locked with LOCK TABLES

事務A獲取到讀鎖之後，在其他事務中是可以正常讀取的，並且也可以再次獲取讀鎖。
在讀鎖中如果想要進行插入操作是不會成功的，出現以下異常：

 Table 'tb_lock_isam' was locked with a READ lock and can't be updated

當前表獲取到讀鎖之後，在當前表沒有釋放讀鎖之前，再獲取寫鎖會一直進入到阻塞狀態。
可以通過非加鎖方式來讀取資料，但是要注意：一定是在不同的事務下

表獨佔寫鎖

WRITE鎖的特性和排它鎖的特性非常相似，都特別霸道：

持有鎖的對談可以讀寫表
只有持有鎖的對談才能存取該表。在釋放鎖之前，沒有其他對談可以存取它
其他對談對錶的鎖請求在WRITE持有鎖時被阻塞

還是通過具體實戰來進行演示效果，關鍵程式碼：lock tables table_name write

session1	session2
select * from tb_lock_isam;	select * from tb_lock_isam;
lock table tb_lock_isam write;
select * from tb_lock_isam;
insert into tb_lock_isam(t_name) values(‘66’);
	select * from tb_lock_isam;
unlock tables;

通過以上實戰，驗證以下結論：

當事務獲取到當前表的WRITE鎖的時候，在當前事務下可以對獲取鎖的表進行任何操作，其他事務無法對錶進行任意操作。
在不同事務下不會對其他表的操作有影響
在當前事務獲取到WRITE鎖之後，只能在當前事務下操作獲取鎖的表，無法操作其他表，否則會出現以下異常

  Table 'tb_index' was not locked with LOCK TABLES'

【注意】

MyISAM在執行查詢語句之前，會自動給涉及的所有表加讀鎖，在執行更新操作前，會自動給涉及的表加寫鎖，這個過程並不需要使用者干預，因此使用者一般不需要使用命令來顯式加鎖

分散式鎖實現

既然已經瞭解到了MySQL鎖相關內容，那麼我們就來看看如何實現，首先我們需要建立一張資料表

當然，只需要初始化建立一次

 create table if not exists fud_distribute_lock(
     id bigint unsigned primary key auto_increment,
     biz varchar(50) comment '業務Key'
     unique(biz)
 ) engine=innodb;

在其中，biz是為了區分不同的業務，也可以理解為資源隔離，並且對biz設定唯一索引，也能夠防止其鎖級別變為表級鎖

既然for udpate就是加鎖成功，事務提交就自動釋放鎖，那麼這個事情就非常好辦了：

 // 省略了構造方法，需要傳入DataSource和biz
 
 private static final String SELECT_SQL = 
     "SELECT * FROM fud_distribute_lock WHERE `biz` = ? for update";
 private static final String INSERT_SQL = 
     "INSERT INTO fud_distribute_lock(`biz`) values(?)";
 
 // 從構造方法中傳入
 private final DataSource source;
 private Connection connection;
 
 public void lock() {
     PreparedStatement psmt = null;
     ResultSet rs = null;
 
     try {
         // while(true); 
         for (; ; ) {
             connection = this.source.getConnection();
             // 關閉自動提交事務
             connection.setAutoCommit(false);
             
             psmt = connection.prepareStatement(SELECT_SQL);
             psmt.setString(1, biz);
             rs = psmt.executeQuery();
             if (rs.next()) {
                 return;
             }
             connection.commit();
             close(connection, psmt, rs);
             // 如果沒有相關查詢，需要插入
             Connection updConnection = this.source.getConnection();
             PreparedStatement insertStatement = null;
             try {
                 insertStatement = updConnection.prepareStatement(INSERT_SQL);
                 insertStatement.setString(1, biz);
                 if (insertStatement.executeUpdate() == 1) {
                     LOGGER.info("建立鎖記錄成功");
                 }
             } catch (Exception e) {
                 LOGGER.error("建立鎖記錄異常：{}", e.getMessage());
             } finally {
                 close(insertStatement, updConnection);
             }
         }
     } catch (Exception e) {
         LOGGER.error("lock異常資訊：{}", e.getMessage());
         throw new BusException(e);
     } finally {
         close(psmt, rs);
     }
 }
 
 public void unlock() {
     try {
         // 事務提交之後自動解鎖
         connection.commit();
         close(connection);
     } catch (Exception e) {
         LOGGER.error("unlock異常資訊：{}", e.getMessage());
         throw new BusException(e);
     }
 }
 
 public void close(AutoCloseable... closeables) {
     Arrays.stream(closeables).forEach(closeable -> {
         if (null != closeable) {
             try {
                 closeable.close();
             } catch (Exception e) {
                 LOGGER.error("close關閉異常：{}", e.getMessage());
             }
         }
     });
 }