字節二面：你有沒有用過分布式鎖？有哪些分布式鎖實現方案？使用分布式鎖有哪些優缺點？

作者：碼農Academy 2024-04-26 08:06:58

分布式鎖是一種在分布式系統中實現互斥控制的機制，確保在多臺機器間，某一資源在同一時刻只被一個服務或者一個請求所訪問或修改。它的核心挑戰在于如何保證在無中心化環境下的全局唯一性和一致性。

引言

隨著業務規模的不斷擴張和技術架構的演進，分布式系統已經成為支撐高并發、海量數據處理的關鍵基礎設施。在分布式環境中，各個節點相對獨立且可能并發地執行任務，這極大地提升了系統的整體性能和可用性。當涉及到對共享資源的訪問和修改時，為了確保數據的一致性和正確性，我們需要一種能在多節點間協調并發操作的技術手段，也就是分布式鎖。

傳統的單機環境下，進程內可以通過本地鎖輕松實現對臨界區資源的互斥訪問。但是，這一方法在分布式系統中不再適用，因為單機鎖無法跨越網絡邊界，無法保證不同節點間的并發控制。分布式鎖正是在這種背景下產生，它是一種能夠實現在分布式系統中多個節點之間協同工作的鎖機制，旨在保護共享資源不受并發沖突的影響，確保在復雜的分布式場景下數據操作的有序性和一致性。

庫存扣減

我們以WMS系統中，訂單出入庫操作庫存為例。

CREATE TABLE `tb_inventory`
(
    `id`                  BIGINT           NOT NULL AUTO_INCREMENT,
    `account_id`          BIGINT           NOT NULL DEFAULT 0 COMMENT '帳套ID',
    `sku`                 VARCHAR(128)     NOT NULL DEFAULT '' COMMENT '商品sku編碼',
    `warehouse_code`      VARCHAR(16)      NOT NULL DEFAULT '' COMMENT '庫存編碼',
    `available_inventory` INT UNSIGNED     NOT NULL DEFAULT 0 COMMENT '可用庫存',
    `create_time`         DATETIME         NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT '創建時間',
    `update_time`         DATETIME         NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT '修改時間',
    `deleted`             TINYINT UNSIGNED NULL     DEFAULT 0 COMMENT '0-未刪除 1/null-已刪除',
    PRIMARY KEY (`id`) USING BTREE,
    UNIQUE KEY uk_warehouse_code (customer_no, warehouse_code, sku, deleted) 
) ENGINE = InnoDB
  AUTO_INCREMENT = 1
  CHARACTER SET = utf8mb4 COMMENT = '庫存表';

庫存表為示例所用，無實際業務參考意義。

關于操作庫存，常見有以下一些錯誤做法：

1、內存中判斷庫存是否充足，并完成扣減

直接在內存中判斷是否有庫存，計算扣減之后的值更新數據庫，并發的情況下會導致庫存相互覆蓋發。

/**
     * 確認訂單出庫
     *
     * @param customerNo
     * @param orderNo
     */
    @Transactional(rollbackFor = Exception.class)
    @Override
    public void confirmOrder(String customerNo, String orderNo) {

        // 查詢訂單信息
        OutboundOrderDO outboundOrderDO = outboundOrderMapper.selectOrderByOrderNo(customerNo, orderNo);
        String warehouseCode = outboundOrderDO.getWarehouseCode();
        // 忽略 訂單信息校驗等，，，
        // 查詢訂單明細  假設我們的出庫訂單是一單一件
        OutboundOrderDetailDO detailDO = orderDetailMapper.selectDetailByOrderNo(outboundOrderDO.getOrderNo());
        String sku = detailDO.getSku();
        Integer qty = detailDO.getQty();

        // 查詢庫存
        TbInventoryDO inventoryDO = tbInventoryMapper.selectSkuInventory(customerNo, warehouseCode, sku);
        Integer availableInventory = inventoryDO.getAvailableInventory();
        // 判斷庫存是否足夠
        if (qty > availableInventory){
            throw new ServiceException(StatusEnum.SERVICE_ERROR, "庫存不足，不能出庫");
        }

        // 剩余庫存
        Integer remainInventory = availableInventory - qty;
        // 扣減庫存
        TbInventoryDO updateInventory = new TbInventoryDO();
        updateInventory.setCustomerNo(customerNo);
        updateInventory.setWarehouseCode(warehouseCode);
        updateInventory.setSku(sku);
        updateInventory.setAvailableInventory(remainInventory);
        tbInventoryMapper.updateInventory(updateInventory);
    }

sql中直接執行更新庫存

<update id="updateInventory">
    UPDATE tb_inventory
    SET available_inventory = #{availableInventory}
    WHERE sku = #{sku}
    AND customer_no = #{customerNo}
    AND warehouse_code = #{warehouseCode}
    AND deleted = 0
</update>

庫存SKU的庫存已經變成了負數：

圖片

2、內存中判斷庫存是否充足，Sql中執行庫存扣減

在InnoDB存儲引擎下，UPDATE通常會應用行鎖，所以在SQL中加入運算避免值的相互覆蓋，但是庫存的數量還是可能變為負數。因為校驗庫存是否充足在內存中執行，并發情況下都會讀到有庫存。

/**
     * 確認訂單出庫
     *
     * @param customerNo
     * @param orderNo
     */
    @Transactional(rollbackFor = Exception.class)
    @Override
    public void confirmOrder(String customerNo, String orderNo) {

        // 查詢訂單信息
        OutboundOrderDO outboundOrderDO = outboundOrderMapper.selectOrderByOrderNo(customerNo, orderNo);
        String warehouseCode = outboundOrderDO.getWarehouseCode();
        // 忽略 訂單信息校驗等，，，
        // 查詢訂單明細  假設我們的出庫訂單是一單一件
        OutboundOrderDetailDO detailDO = orderDetailMapper.selectDetailByOrderNo(outboundOrderDO.getOrderNo());
        String sku = detailDO.getSku();
        Integer qty = detailDO.getQty();

        // 查詢庫存
        TbInventoryDO inventoryDO = tbInventoryMapper.selectSkuInventory(customerNo, warehouseCode, sku);
        Integer availableInventory = inventoryDO.getAvailableInventory();
        // 判斷庫存是否足夠
        if (qty > availableInventory){
            throw new ServiceException(StatusEnum.SERVICE_ERROR, "庫存不足，不能出庫");
        }

        // 扣減庫存
        TbInventoryDO updateInventory = new TbInventoryDO();
        updateInventory.setCustomerNo(customerNo);
        updateInventory.setWarehouseCode(warehouseCode);
        updateInventory.setSku(sku);
        // 庫存差值
        updateInventory.setDiffInventory(qty);
        tbInventoryMapper.updateInventory(updateInventory);
    }

庫存扣減在sql中進行

<update id="updateInventory">
    UPDATE tb_inventory
    SET available_inventory = available_inventory - #{diffInventory}
    WHERE sku = #{sku}
    AND customer_no = #{customerNo}
    AND warehouse_code = #{warehouseCode}
    AND deleted = 0
  </update>

庫存SKU的庫存已經變成了負數：

圖片

在操作庫存方法上使用synchronized

雖然synchronized可以防止在多并發環境下，多個線程并發訪問這個庫存操作方法，但是synchronized的作用在方法結束之后就失效了，可能此時事務并沒有提交，導致可能其他的線程會在拿到鎖之后讀取到舊庫存數據，在執行扣除時，依然可能會造成庫存扣減不對。

/**
     * 確認訂單出庫
     *
     * @param customerNo
     * @param orderNo
     */
    @Transactional(rollbackFor = Exception.class)
    @Override
    public synchronized void confirmOrder(String customerNo, String orderNo) {

        // 查詢訂單信息
        OutboundOrderDO outboundOrderDO = outboundOrderMapper.selectOrderByOrderNo(customerNo, orderNo);
        String warehouseCode = outboundOrderDO.getWarehouseCode();
        // 忽略 訂單信息校驗等，，，
        // 查詢訂單明細  假設我們的出庫訂單是一單一件
        OutboundOrderDetailDO detailDO = orderDetailMapper.selectDetailByOrderNo(outboundOrderDO.getOrderNo());
        String sku = detailDO.getSku();
        Integer qty = detailDO.getQty();

        // 查詢庫存
        TbInventoryDO inventoryDO = tbInventoryMapper.selectSkuInventory(customerNo, warehouseCode, sku);
        Integer availableInventory = inventoryDO.getAvailableInventory();
        // 判斷庫存是否足夠
        if (qty > availableInventory){
            throw new ServiceException(StatusEnum.SERVICE_ERROR, "庫存不足，不能出庫");
        }

        // 扣減庫存
        TbInventoryDO updateInventory = new TbInventoryDO();
        updateInventory.setCustomerNo(customerNo);
        updateInventory.setWarehouseCode(warehouseCode);
        updateInventory.setSku(sku);
        // 庫存差值
        updateInventory.setDiffInventory(qty);
        tbInventoryMapper.updateInventory(updateInventory);
    }

庫存SKU的庫存已經變成了負數：

圖片

從上面的錯誤案例來看，在操作庫存時，不是原子性的，導致庫存操作失敗。以下我們從單體以及分布式系統兩個方向探討如何保證數據的一致性和正確性。

單機系統

在單機系統中，數據和業務邏輯都集中在一個進程中，面對并發訪問共享資源的情況，需要依靠鎖機制和數據庫的事務管理（行鎖）來維護數據的正確性和一致性。

對于鎖機制，我們不管是采用synchronized還是Lock等，我們要保證的一個條件就是：要讓數據庫的事務在鎖的控制范圍之內。

針對上述錯誤案例，我們可以將鎖作用于事務之外，即將鎖放在庫存操作方法的上一層（例如service層）。

@Service
public class OrderServiceImpl implements IOrderService {

    private IOrderManager orderManager;

    /**
     * 確認訂單出庫
     *
     * @param customerNo
     * @param orderNo
     */
    @Override
    public synchronized void confirmOrder(String customerNo, String orderNo) {
        orderManager.confirmOrder(customerNo, orderNo);
    }

    @Autowired
    public void setOrderManager(IOrderManager orderManager) {
        this.orderManager = orderManager;
    }
}

此時我們在操作庫存，會因為庫存不夠，導致庫存操作失敗：

圖片

這種方式雖然可以實現數據一致性和正確性，但是并不是很推薦，因為我們的事務要控制的粒度盡可能的小。

推薦的方式，是我們再鎖的控制范圍去提交事務。即手動提交事務。使用TransactionTemplate或直接在代碼中調用PlatformTransactionManager的getTransaction和commit方法來手動管理事務。

@Autowired
    private PlatformTransactionManager transactionManager;


    /**
     * 確認訂單出庫
     *
     * @param customerNo
     * @param orderNo
     */
    @Override
    public synchronized void confirmOrder(String customerNo, String orderNo) {
        // 查詢訂單信息
        OutboundOrderDO outboundOrderDO = outboundOrderMapper.selectOrderByOrderNo(customerNo, orderNo);
        String warehouseCode = outboundOrderDO.getWarehouseCode();

        TransactionStatus status = transactionManager.getTransaction(new DefaultTransactionDefinition());
        // 忽略 訂單信息校驗等，，，
        // 查詢訂單明細  假設我們的出庫訂單是一單一件
        OutboundOrderDetailDO detailDO = orderDetailMapper.selectDetailByOrderNo(outboundOrderDO.getOrderNo());
        String sku = detailDO.getSku();
        Integer qty = detailDO.getQty();

        // 查詢庫存
        TbInventoryDO inventoryDO = tbInventoryMapper.selectSkuInventory(customerNo, warehouseCode, sku);
        Integer availableInventory = inventoryDO.getAvailableInventory();
        // 判斷庫存是否足夠
        if (qty > availableInventory){
            System.err.println("庫存不足，不能出庫");
            throw new ServiceException(StatusEnum.SERVICE_ERROR, "庫存不足，不能出庫");
        }

        // 扣減庫存
        TbInventoryDO updateInventory = new TbInventoryDO();
        updateInventory.setCustomerNo(customerNo);
        updateInventory.setWarehouseCode(warehouseCode);
        updateInventory.setSku(sku);
        // 庫存差值
        updateInventory.setDiffInventory(qty);
        tbInventoryMapper.updateInventory(updateInventory);
        // 提交事務
        transactionManager.commit(status);
    }

此時我們再去執庫存操作，會因為庫存不夠，導致庫存操作失敗：

圖片

對于上述同步鎖的實現，我們最好使用Lock得方式去實現，可以更精細控制同步邏輯。

@Autowired
private PlatformTransactionManager transactionManager;

private final Lock orderLock = new ReentrantLock();
/**
 * 確認訂單出庫
 *
 * @param customerNo
 * @param orderNo
 */
@Override
public void confirmOrder(String customerNo, String orderNo) {
    // 查詢訂單信息
    OutboundOrderDO outboundOrderDO = outboundOrderMapper.selectOrderByOrderNo(customerNo, orderNo);
    String warehouseCode = outboundOrderDO.getWarehouseCode();

    try {
        // 嘗試獲取鎖，最多等待timeout時間
        if (orderLock.tryLock(1, TimeUnit.SECONDS)) {
            // 成功獲取到鎖，執行確認訂單的邏輯
            TransactionStatus status = transactionManager.getTransaction(new DefaultTransactionDefinition());
            try {
                // 忽略 訂單信息校驗等，，，
                // 查詢訂單明細  假設我們的出庫訂單是一單一件
                OutboundOrderDetailDO detailDO = orderDetailMapper.selectDetailByOrderNo(outboundOrderDO.getOrderNo());
                String sku = detailDO.getSku();
                Integer qty = detailDO.getQty();

                // 查詢庫存
                TbInventoryDO inventoryDO = tbInventoryMapper.selectSkuInventory(customerNo, warehouseCode, sku);
                Integer availableInventory = inventoryDO.getAvailableInventory();
                // 判斷庫存是否足夠
                if (qty > availableInventory){
                    System.err.println("庫存不足，不能出庫");
                    throw new ServiceException(StatusEnum.SERVICE_ERROR, "庫存不足，不能出庫");
                }

                // 扣減庫存
                TbInventoryDO updateInventory = new TbInventoryDO();
                updateInventory.setCustomerNo(customerNo);
                updateInventory.setWarehouseCode(warehouseCode);
                updateInventory.setSku(sku);
                // 庫存差值
                updateInventory.setDiffInventory(qty);
                tbInventoryMapper.updateInventory(updateInventory);
                // 提交事務
                transactionManager.commit(status);
            }catch (Exception e){
                // 回滾事務
                transactionManager.rollback(status);
                // 處理異常
                e.printStackTrace();
            }finally {
                // 釋放鎖
                orderLock.unlock();
            }
        } else {
            // 獲取鎖超時
            System.out.println("Failed to confirm order within the timeout period: " +orderNo);
            // 處理超時情況，比如記錄日志、通知用戶等
        }
    } catch (InterruptedException e) {
        // 如果在等待鎖的過程中線程被中斷，處理中斷異常
        Thread.currentThread().interrupt();
        // ... 處理中斷邏輯 ...
    }
        
}

在單機系統中，上述方法可以保證數據一致性以及正確性，但是實際業務中，我們應用通常都部署在多個服務器中，此時上述方案就不能保證了，就需要分布式鎖來解決了。

分布式鎖的實現

在單機系統中，鎖是一種基本的同步機制，用于控制多個線程對共享資源的并發訪問。當我們升級到分布式系統時，由于服務分散在多個節點之上，原本在單機環境下使用的鎖機制無法直接跨越多個節點來協調資源訪問。所以此時，分布式鎖作為一種擴展的鎖概念應運而生。分布式鎖是一種跨多個節點、進程或服務的同步原語，它允許在分布式系統中協調對共享資源的訪問，確保在任何時候只有一個節點能夠獨占地執行操作，即使這些節點分布在不同的物理或虛擬機器上。

分布式鎖的基本要素

1. 互斥性：這是分布式鎖最基本的要求，意味著在任意時刻，只有一個客戶端（無論是進程、線程還是服務實例）能夠持有并使用鎖，從而確保共享資源不會同時被多個客戶端修改。

2. 持久性：分布式鎖必須具備一定的持久化能力，即便服務重啟或網絡短暫斷開，鎖的狀態仍然能夠得到保持。

3. 可重入性：類似于單機環境下的可重入鎖，分布式鎖也應該支持同一客戶端在持有鎖的同時再次請求鎖而不被阻塞，這對于遞歸調用或涉及多個資源訪問的操作至關重要。

4. 公平性（Fairness）：在某些場景下，要求鎖分配遵循一定的公平原則，即等待最久的客戶端在鎖釋放時優先獲得鎖。雖然不是所有分布式鎖實現都需要考慮公平性，但在某些高性能或高并發的系統中，公平性是非常重要的。

5. 容錯性：分布式鎖服務應當具備一定的容錯能力，即即使一部分服務節點發生故障，仍能保證鎖功能的正確運行，防止死鎖和數據不一致。這通常通過服務冗余和復制機制來實現，如使用Raft、Paxos等一致性協議或基于ZooKeeper、etcd等分布式協調服務。

常見分布式鎖解決方案

基于數據庫實現

1.數據庫悲觀鎖

悲觀鎖以預防性策略處理并發沖突，它假設并發訪問導致的數據沖突是常態。因此，在訪問數據之前，它會積極地獲取并持有鎖，確保在鎖未釋放時，其他事務無法對同一數據進行訪問。通過運用SELECT ... FOR UPDATE SQL語句，能夠在查詢階段即鎖定相關行，實現數據的獨占訪問。然而，重要的是要注意，此操作應僅針對唯一鍵執行，否則可能會大幅增加鎖定范圍和潛在的鎖表風險，從而影響系統的并發性能與效率。

最常見的做法是直接在業務數據上使用SELECT ... FOR UPDATE，例如：

<select id="selectSkuInventoryForUpdate" resultType="com.springboot.mybatis.entity.TbInventoryDO">
    SELECT *
    FROM tb_inventory
    WHERE sku = #{sku}
    AND customer_no = #{customerNo}
    AND warehouse_code = #{warehouseCode}
    AND deleted = 0
    FOR UPDATE
  </select>

在一個事務中，先使用SELECT ... FOR UPDATE后，在執行更新。

/**
     * 使用SELECT... FOR UPDATE 實現分布式鎖，扣減庫存
     * @param customerNo
     * @param orderNo
     */
    @Transactional(rollbackFor = Exception.class)
    @Override
    public void confirmOrderWithLock(String customerNo, String orderNo) {
        // 查詢訂單信息
        OutboundOrderDO outboundOrderDO = outboundOrderMapper.selectOrderByOrderNo(customerNo, orderNo);
        String warehouseCode = outboundOrderDO.getWarehouseCode();

        // 查詢訂單明細  假設我們的出庫訂單是一單一件
        OutboundOrderDetailDO detailDO = orderDetailMapper.selectDetailByOrderNo(outboundOrderDO.getOrderNo());
        String sku = detailDO.getSku();
        Integer qty = detailDO.getQty();

        // 查詢庫存
        TbInventoryDO inventoryDO = tbInventoryMapper.selectSkuInventoryForUpdate(customerNo, warehouseCode, sku);
        Integer availableInventory = inventoryDO.getAvailableInventory();
        // 判斷庫存是否足夠
        if (qty > availableInventory){
            System.err.println("庫存不足，不能出庫");
            throw new ServiceException(StatusEnum.SERVICE_ERROR, "庫存不足，不能出庫");
        }

        // 扣減庫存
        TbInventoryDO updateInventory = new TbInventoryDO();
        updateInventory.setCustomerNo(customerNo);
        updateInventory.setWarehouseCode(warehouseCode);
        updateInventory.setSku(sku);
        // 庫存差值
        updateInventory.setDiffInventory(qty);
        tbInventoryMapper.updateInventory(updateInventory);
    }

但是，這種實現方式，很容易造成業務表的鎖壓力，特別是數據量大，并發量高的時候。所以，還有一種做法是，專門維護一張鎖的表，而不是直接在業務數據表上使用SELECT FOR UPDATE。這種方式在某些場景下可以幫助簡化鎖的管理，并且可以在一定程度上減輕對業務數據表的鎖定壓力。（其實實現方式，類似Redis實現的分布式鎖，只是用數據庫實現了而已）。其實現流程，如下：

數據庫實現悲觀鎖流程

1. 創建鎖表：首先，創建一張鎖表，例如lock_table，包含lock_key（用于標識需要鎖定的業務資源）、lock_holder（持有鎖的客戶端標識，如用戶ID或事務ID）、acquire_time（獲取鎖的時間）等字段。

CREATE TABLE `tb_lock`
(
    id           BIGINT AUTO_INCREMENT
        PRIMARY KEY,
    lock_key     VARCHAR(255)                               NOT NULL DEFAULT '' COMMENT '鎖的業務編碼。對應業務表的唯一鍵',
    lock_holder  VARCHAR(32)                                NOT NULL DEFAULT '' COMMENT '持有鎖的客戶端標識',
    acquire_time DATETIME                                   NOT NULL COMMENT '獲取鎖的時間',
    create_time  DATETIME         DEFAULT CURRENT_TIMESTAMP NOT NULL COMMENT '創建時間',
    update_time  DATETIME         DEFAULT CURRENT_TIMESTAMP NOT NULL ON UPDATE CURRENT_TIMESTAMP COMMENT '修改時間',
    deleted      TINYINT UNSIGNED DEFAULT '0'               NULL COMMENT '0-未刪除 1/null-已刪除',
    UNIQUE KEY uk_lock (lock_key, deleted)
) ENGINE = InnoDB
  AUTO_INCREMENT = 1
  CHARACTER SET = utf8mb4 COMMENT = 'Lock表';

插入鎖記錄：當客戶端想要獲取鎖時，嘗試在lock_table中插入一條記錄，其中lock_key對應需要保護的業務資源，例如商品SKU。插入操作通常是通過INSERT INTO ... ON DUPLICATE KEY UPDATE這樣的語句實現，以確保在存在相同鎖鍵的情況下更新記錄，否則插入新記錄，這一步相當于獲取鎖。

<insert id="insertLock">
    INSERT INTO tb_lock
    (lock_key,lock_holder,acquire_time)
    VALUES
    (#{lockKey},#{lockHolder},#{acquireTime})
  </insert>

使用 SELECT FOR UPDATE：在插入鎖記錄時，可以通過SELECT ... FOR UPDATE鎖定鎖表中的相應記錄，確保在當前事務結束前，其他事務無法更新或刪除這條鎖記錄。

<select id="selectLockByLockKey" resultType="com.springboot.mybatis.entity.TbLockDO">
    SELECT *
    FROM tb_lock
    WHERE lock_key = #{lockKey} AND deleted = 0
    FOR UPDATE
</select>

4. 檢查鎖狀態：在獲取鎖時，可以檢查鎖是否已被持有，比如檢查lock_holder字段，如果已有其他事務持有鎖，則獲取鎖失敗，需要等待或重試。

// 嘗試獲取鎖
tryLock(lockKey, lockHolder);
// 使用SELECT FOR UPDATE鎖定鎖表記錄
TbLockDO tbLockDO = tbLockMapper.selectLockByLockKey(lockKey);
if (!tbLockDO.getLockHolder().equals(lockHolder)) {
    // 鎖已被其他客戶端持有，獲取鎖失敗，需要處理此異常情況
    throw new IllegalStateException("Lock is held by another client.");
}

釋放鎖：當業務操作完成時，可以通過刪除或更新鎖表中的對應記錄來釋放鎖。

<delete id="deleteLockByLockKey" parameterType="java.lang.String">
    DELETE FROM tb_lock
    WHERE lock_key = #{lockKey}
    AND lock_holder = #{lockHolder}
    AND deleted = 0
</delete>

基于數據庫悲觀鎖實現，代碼如下：

@Transactional(rollbackFor = Exception.class)
    @Override
    public void confirmOrderWithLock(String customerNo, String orderNo) {
        // 查詢訂單信息
        OutboundOrderDO outboundOrderDO = outboundOrderMapper.selectOrderByOrderNo(customerNo, orderNo);
        String warehouseCode = outboundOrderDO.getWarehouseCode();

        // 查詢訂單明細  假設我們的出庫訂單是一單一件
        OutboundOrderDetailDO detailDO = orderDetailMapper.selectDetailByOrderNo(outboundOrderDO.getOrderNo());
        String sku = detailDO.getSku();
        Integer qty = detailDO.getQty();

        String lockKey = String.format("inventory:%s_%s_%s", customerNo, warehouseCode, sku);
        String lockHolder = Thread.currentThread().getName();
        try {
            // 嘗試獲取鎖
            tryLock(lockKey, lockHolder);
            // 使用SELECT FOR UPDATE鎖定鎖表記錄
            TbLockDO tbLockDO = tbLockMapper.selectLockByLockKey(lockKey);
            if (!tbLockDO.getLockHolder().equals(lockHolder)) {
                // 鎖已被其他客戶端持有，獲取鎖失敗，需要處理此異常情況
                throw new IllegalStateException("Lock is held by another client.");
            }
            // 查詢庫存
            TbInventoryDO inventoryDO = tbInventoryMapper.selectSkuInventoryForUpdate(customerNo, warehouseCode, sku);
            Integer availableInventory = inventoryDO.getAvailableInventory();
            // 判斷庫存是否足夠
            if (qty > availableInventory){
                System.err.println("庫存不足，不能出庫");
                throw new ServiceException(StatusEnum.SERVICE_ERROR, "庫存不足，不能出庫");
            }

            // 扣減庫存
            TbInventoryDO updateInventory = new TbInventoryDO();
            updateInventory.setCustomerNo(customerNo);
            updateInventory.setWarehouseCode(warehouseCode);
            updateInventory.setSku(sku);
            // 庫存差值
            updateInventory.setDiffInventory(qty);
            tbInventoryMapper.updateInventory(updateInventory);
        }finally {
            unlock(lockKey, lockHolder);
        }
    }


    /**
     * 嘗試獲取鎖
     * @param lockKey 鎖的key 業務編碼
     * @param lockHolder 鎖的持有者
     * @return 是否獲取成功
     */
    private void tryLock(String lockKey, String lockHolder) {
        TbLockDO tbLockDO = new TbLockDO();
        tbLockDO.setLockKey(lockKey);
        tbLockDO.setLockHolder(lockHolder);
        tbLockDO.setAcquireTime(LocalDateTime.now());
        //插入一條數據   insert into
        tbLockMapper.insertLock(tbLockDO);
    }

    /**
     * 鎖釋放
     * @param lockKey 鎖的key 業務編碼
     */
    private void unlock(String lockKey, String lockHolder){
        tbLockMapper.deleteLockByLockKey(lockKey, lockHolder);
    }

圖片

數據庫悲觀鎖實現分布式鎖可以防止并發沖突，確保在事務結束前，這些記錄不會被其他并發事務修改。它還可以控制鎖的粒度，提供行級別的鎖定，減少鎖定范圍，提高并發性能。這種方式非常適合于處理需要更新的事務場景，特別是銀行轉賬、庫存扣減等需要保證數據完整性和一致性的操作。

但是，需要注意的是，過度或不當使用SELECT FOR UPDATE會導致更多的行被鎖定，在高并發場景下，如果大量事務都在等待獲取鎖，可能會導致鎖等待和死鎖問題，并且當事務持有SELECT FOR UPDATE的鎖時，其他事務嘗試修改這些鎖定的行會陷入等待狀態，直至鎖釋放。這可能導致其他事務的延遲和系統吞吐量下降，長時間持有鎖會導致數據庫資源（如內存、連接數等）消耗增大，特別是長事務中持有鎖時間較長，會影響系統的總體性能。所以我們在使用時要特別注意不要再長事務中使用悲觀鎖。

2.數據庫樂觀鎖

樂觀鎖假定并發沖突不太可能發生，因此在讀取數據時不鎖定資源，而是在更新數據時驗證數據是否被其他事務修改過。

在數據庫表中添加一個version字段。

ALTER TABLE `tb_inventory` ADD COLUMN `version` INT NOT NULL DEFAULT 0 COMMENT '樂觀鎖版本' AFTER available_inventory;

每次更新時將version字段加1。在更新數據時，通過UPDATE語句附帶WHERE version = oldVersion條件，只有當version值不變時更新操作才會成功。若version已變，則表示數據已被其他事務修改，此次更新失敗。

<update id="updateInventorWithVersion">
    UPDATE tb_inventory
    SET available_inventory = available_inventory - #{diffInventory},
        version = #{version} + 1
    WHERE sku = #{sku}
    AND customer_no = #{customerNo}
    AND warehouse_code = #{warehouseCode}
    AND version = #{version}
    AND deleted = 0
  </update>

基于樂觀鎖實現的方案：

@Transactional(rollbackFor = Exception.class)
@Override
public void confirmOrderWithVersion(String customerNo, String orderNo) {
    // 查詢訂單信息
    OutboundOrderDO outboundOrderDO = outboundOrderMapper.selectOrderByOrderNo(customerNo, orderNo);
    String warehouseCode = outboundOrderDO.getWarehouseCode();

    // 查詢訂單明細  假設我們的出庫訂單是一單一件
    OutboundOrderDetailDO detailDO = orderDetailMapper.selectDetailByOrderNo(outboundOrderDO.getOrderNo());
    String sku = detailDO.getSku();
    Integer qty = detailDO.getQty();

    // 查詢庫存
    TbInventoryDO inventoryDO = tbInventoryMapper.selectSkuInventory(customerNo, warehouseCode, sku);
    Integer availableInventory = inventoryDO.getAvailableInventory();
    Integer curVersion = inventoryDO.getVersion();
    // 判斷庫存是否足夠
    if (qty > availableInventory){
        System.err.println("庫存不足，不能出庫");
        throw new ServiceException(StatusEnum.SERVICE_ERROR, "庫存不足，不能出庫");
    }

    // 扣減庫存
    TbInventoryDO updateInventory = new TbInventoryDO();
    updateInventory.setCustomerNo(customerNo);
    updateInventory.setWarehouseCode(warehouseCode);
    updateInventory.setSku(sku);
    // 設置當前數據版本號
    updateInventory.setVersion(curVersion);
    // 庫存差值
    updateInventory.setDiffInventory(qty);
    updateInventory.setVersion(inventoryDO.getVersion());
    int updateRows = tbInventoryMapper.updateInventorWithVersion(updateInventory);
    if (updateRows != 1){
        System.err.println("更新庫存時發生并發沖突，請重試");
        throw new ServiceException(StatusEnum.SERVICE_ERROR, "更新庫存時發生并發沖突，請重試");
    }
}

圖片

樂觀鎖假定大多數情況下不會有并發沖突，所以在讀取數據時不立即加鎖，而是等到更新數據時才去檢查是否有其他事務進行了改動，這樣可以減少鎖的持有時間，提高了系統的并發性能。并且，樂觀鎖在數據更新時才檢查沖突，而不是在獲取數據時就加鎖，所以大大降低了死鎖的風險。并且因為不常加鎖，所以減少了數據庫級別的鎖管理開銷，非常適合對于讀多寫少的場景。

但是，當并發寫入較多時，可能出現大量更新沖突，需要不斷地重試事務以獲得成功的更新。過多的重試可能導致性能下降，特別是在并發度極高時，可能會形成“ABA”問題。并且在極端并發條件下，如果沒有正確的重試機制或超時機制，樂觀鎖可能無法保證強一致性。尤其是在涉及多個表的復雜事務中，單個樂觀鎖可能不足以解決所有并發問題。

基于Redis實現

1.Redis的setNX實現

Redis的setNX（set if not exists）命令是原子操作，當鍵不存在時才設置值，設置成功則返回true，否則返回false。通過這個命令可以快速地在Redis中爭奪一把鎖。

利用Redis，我們可以生成一個唯一的鎖ID作為key的一部分。然后使用setNX嘗試設置key-value對，value可以是過期時間戳。若設置成功，則認為獲取鎖成功，執行業務邏輯。在業務邏輯完成后，刪除對應key釋放鎖，或設置過期時間自動釋放。

@Slf4j
public class RedisDistributedLock implements AutoCloseable{

    private final StringRedisTemplate stringRedisTemplate;
    private final DefaultRedisScript<Boolean> unlockScript;

    /**鎖的key*/
    private final String lockKey;
    /**鎖過期時間*/
    private final Integer expireTime;

    private static final String UNLOCK_LUA_SCRIPT = "if redis.call(\"get\", KEYS[1]) == ARGV[1] then\n" +
                                                    "    return redis.call(\"del\", KEYS[1])\n" +
                                                    "else\n" +
                                                    "    return 0\n" +
                                                    "end";
    public RedisDistributedLock(StringRedisTemplate stringRedisTemplate, String lockKey, Integer expireTime) {
        this.stringRedisTemplate = stringRedisTemplate;
        this.lockKey = lockKey;
        this.expireTime = expireTime;
        // 初始化Lua解鎖腳本
        this.unlockScript = new DefaultRedisScript<>();
        unlockScript.setScriptText(UNLOCK_LUA_SCRIPT);
        unlockScript.setResultType(Boolean.class);
    }

    /**
     * 獲取鎖
     * @return 是否獲取成功
     */
    public Boolean getLock() {
        String value = UUID.randomUUID().toString();
        try {
            return stringRedisTemplate.opsForValue().setIfAbsent(lockKey, value, expireTime, TimeUnit.SECONDS);
        } catch (Exception e) {
            log.error("獲取分布式鎖失敗: {}", e.getMessage());
            return false;
        }
    }

    /**
     * 釋放鎖
     * @return 是否釋放成功
     */
    public Boolean unLock() {
        // 使用Lua腳本進行解鎖操作
        List<String> keys = Collections.singletonList(lockKey);
        Object result = stringRedisTemplate.execute(unlockScript, keys, stringRedisTemplate.opsForValue().get(lockKey));
        boolean unlocked = (Boolean) result;
        log.info("釋放鎖的結果: {}", unlocked);
        return unlocked;
    }


    @Override
    public void close() throws Exception {
        unLock();
    }
}

然后，我們在處理庫存時，先嘗試獲取鎖，如果獲取到鎖，則就可以更新庫存。

@Transactional(rollbackFor = Exception.class)
    @Override
    public void confirmOrderWithRedisNx(String customerNo, String orderNo) {
        // 查詢訂單信息
        OutboundOrderDO outboundOrderDO = outboundOrderMapper.selectOrderByOrderNo(customerNo, orderNo);
        String warehouseCode = outboundOrderDO.getWarehouseCode();

        // 查詢訂單明細  假設我們的出庫訂單是一單一件
        OutboundOrderDetailDO detailDO = orderDetailMapper.selectDetailByOrderNo(outboundOrderDO.getOrderNo());
        String sku = detailDO.getSku();
        Integer qty = detailDO.getQty();

        String lockKey = String.format("inventory:%s_%s_%s", customerNo, warehouseCode, sku);
        // 30秒過期
        try (RedisDistributedLock lock = new RedisDistributedLock(stringRedisTemplate, lockKey, 30)) {
            if (lock.getLock()) {
                // 查詢庫存
                TbInventoryDO inventoryDO = tbInventoryMapper.selectSkuInventory(customerNo, warehouseCode, sku);
                Integer availableInventory = inventoryDO.getAvailableInventory();
                // 判斷庫存是否足夠
                if (qty > availableInventory){
                    System.err.println("庫存不足，不能出庫");
                    throw new ServiceException(StatusEnum.SERVICE_ERROR, "庫存不足，不能出庫");
                }

                // 扣減庫存
                TbInventoryDO updateInventory = new TbInventoryDO();
                updateInventory.setCustomerNo(customerNo);
                updateInventory.setWarehouseCode(warehouseCode);
                updateInventory.setSku(sku);
                // 庫存差值
                updateInventory.setDiffInventory(qty);
                tbInventoryMapper.updateInventory(updateInventory);
            } else {
                log.error("更新庫存時發生并發沖突，請重試");
                throw new ServiceException(StatusEnum.SERVICE_ERROR, "更新庫存時發生并發沖突，請重試");
            }
        } catch (Exception e) {
            log.error("處理分布式鎖時發生錯誤: {}", e.getMessage());
        }
    }

圖片

Redis作為內存數據庫，其操作速度快，setNX的執行時間幾乎可以忽略不計，尤其適合高并發場景下的鎖請求。Redis作為一個可以獨立的服務，可以輕松實現不同進程或服務器之間的互斥鎖。而setNX命令是原子操作，能夠在Redis這一單線程環境下以原子性的方式實現鎖的獲取，簡單一行命令即可實現鎖的爭搶。同時可以通過EX或PX參數，可以在設置鎖時一并設定過期時間，避免因意外情況導致的死鎖。

但是單純使用setNX并不能自動續期，一旦鎖過期而又未主動釋放，可能出現鎖被其他客戶端誤獲取的情況，需要額外實現鎖的自動續期機制，例如使用WATCH和MULTI命令組合，或者SET命令的新參數如SET key value PX milliseconds NX XX。而setNX在獲取不到鎖時會立即返回失敗，所以我們必須輪詢或使用某種延時重試策略來不斷嘗試獲取鎖。并且如果多個客戶端同時請求鎖，Redis并不會保證特定的排隊順序，可能導致“饑餓”現象（即某些客戶端始終無法獲取鎖）。

雖然Redis的setNX命令在實現分布式鎖方面提供了便捷性和高性能，但要構建健壯、可靠的分布式鎖解決方案，往往還需要結合其他命令（如expire、watch、multi/exec等）以及考慮到各種邊緣情況和容錯機制。一些成熟的Redis客戶端庫（如Redisson、Jedis）提供了封裝好的分布式鎖實現，解決了上述許多問題。

基于Redisson實現

Redisson是一個高性能、開源的Java駐內存數據網格，它基于Redis，并提供了眾多分布式數據結構和一套分布式服務，例如分布式鎖、信號量、閉鎖、隊列、映射等。Redisson使得開發者能夠更容易地在Java應用程序中使用Redis，特別是對分布式環境下的同步原語提供了豐富的API支持。

Redisson的分布式鎖核心原理基于Redis命令，但進行了增強和封裝，提供了一種更加可靠和易于使用的分布式鎖實現。他實現分布式鎖的思路與Redis的setNx實現類似。但是，相比較與Redis的setNx實現分布式鎖，Redisson還支持可重入鎖，即同一個線程在已經獲得鎖的情況下可以再次獲取鎖而不被阻塞。內部通過計數器記錄持有鎖的次數，每次成功獲取鎖時計數器遞增，釋放鎖時遞減，只有當計數器歸零時才真正釋放鎖。Redisson使用了看門狗（Watchdog）機制來監控鎖的狀態，定期自動延長鎖的有效期，這樣即使持有鎖的客戶端暫時凍結或網絡抖動，鎖也不會因為超時而被提前釋放。并且，對于Redis集群，Redisson還可以實現RedLock算法，通過在多個Redis節點上分別獲取鎖，增加分布式鎖的可用性和容錯能力。

我們使用Redisson實現分布式鎖，實現庫存扣減。

<dependency>
    <groupId>org.redisson</groupId>
    <artifactId>redisson-spring-boot-starter</artifactId>
    <version>3.17.7</version>
</dependency>

spring:
  redisson:
    address: "redis://127.0.0.1:6379"
    password:

@Override
public void confirmOrderWithRedisson(String customerNo, String orderNo) {
        // 查詢訂單信息
        OutboundOrderDO outboundOrderDO = outboundOrderMapper.selectOrderByOrderNo(customerNo, orderNo);
        String warehouseCode = outboundOrderDO.getWarehouseCode();

        // 查詢訂單明細  假設我們的出庫訂單是一單一件
        OutboundOrderDetailDO detailDO = orderDetailMapper.selectDetailByOrderNo(outboundOrderDO.getOrderNo());
        String sku = detailDO.getSku();
        Integer qty = detailDO.getQty();

        String lockKey = String.format("inventory:%s_%s_%s", customerNo, warehouseCode, sku);
        // 30秒過期
        RLock lock = redissonClient.getLock(lockKey);
        try {
            if (lock.tryLock(30, TimeUnit.SECONDS)) {
                // 查詢庫存
                TbInventoryDO inventoryDO = tbInventoryMapper.selectSkuInventory(customerNo, warehouseCode, sku);
                Integer availableInventory = inventoryDO.getAvailableInventory();
                // 判斷庫存是否足夠
                if (qty > availableInventory){
                    System.err.println("庫存不足，不能出庫");
                    throw new ServiceException(StatusEnum.SERVICE_ERROR, "庫存不足，不能出庫");
                }

                // 扣減庫存
                TbInventoryDO updateInventory = new TbInventoryDO();
                updateInventory.setCustomerNo(customerNo);
                updateInventory.setWarehouseCode(warehouseCode);
                updateInventory.setSku(sku);
                // 庫存差值
                updateInventory.setDiffInventory(qty);
                tbInventoryMapper.updateInventory(updateInventory);
            } else {
                log.error("更新庫存時發生并發沖突，請重試");
                throw new ServiceException(StatusEnum.SERVICE_ERROR, "更新庫存時發生并發沖突，請重試");
            }
        }catch (Exception e){
            throw new ServiceException(StatusEnum.SERVICE_ERROR, "獲取分布式鎖時被中斷");
        }finally {
            // 無論成功與否，都要釋放鎖
            if (lock.isLocked() && lock.isHeldByCurrentThread()) {
                lock.unlock();
            }
        }
    }

圖片

Redisson支持多種類型的分布式鎖，包括可重入鎖（RLock）、讀寫鎖（RReadWriteLock）、公平鎖（RFairLock）等，滿足不同業務場景的需求。Redisson支持鎖的自動續期功能，可以防止因為鎖持有者在業務處理過程中長時間未完成而導致鎖過期被其他客戶端獲取。對于Redisson RedLock算法（多節點部署時），即使部分Redis節點失效，也能在大多數Redis節點存活的情況下維持鎖的穩定性，增強了系統的容錯性和高可用性。

相較于簡單的數據庫悲觀鎖，Redisson的分布式鎖實現更為復雜。雖然Redisson提供了自動續期機制，但如果客戶端在獲取鎖后突然崩潰且沒有正常釋放鎖，理論上仍然有可能導致鎖泄漏。雖然Redisson也提供了超時設置，但極端情況下仍需結人工清理機制或者其他的方案來預防此類問題。

使用Zookeeper

在Zookeeper中實現分布式鎖的基本原理是利用Zookeeper的臨時節點和Watcher監聽機制。

客戶端在Zookeeper中指定的某個路徑下創建臨時有序節點，每個節點名稱后都會附加一個唯一的遞增數字，表示節點的順序。當多個客戶端同時請求鎖時，它們都會創建各自的臨時有序節點。

客戶端按照節點順序判斷自己是否可以獲得鎖。節點順序最小的客戶端被認為是鎖的持有者，它觀察到的序號比自己大的所有節點都是待解鎖的隊列。鎖的持有者繼續執行業務邏輯，其它客戶端則會注冊Watcher監聽比自己序號小的那個節點。

當鎖持有者完成業務處理后，會刪除它創建的臨時節點，Zookeeper會觸發Watcher通知等待隊列中的下一個節點。接收到通知的下一個節點發現其觀察的節點已刪除，于是重新檢查當前路徑下剩余節點的順序，如果自己是現在最小的節點，則認為獲得了鎖。

Watcher機制允許客戶端監聽Zookeeper上的節點變化事件，當節點被創建、刪除、更新時，Zookeeper會向注冊了相應事件的客戶端發送通知。在分布式鎖場景中，客戶端通過注冊Watcher來監聽鎖持有者的節點狀態，以便在鎖釋放時及時獲取鎖。

圖片

而我們使用Apache Curator框架作為Zookeeper客戶端實現分布式鎖。Curator擁有良好的架構設計，提供了豐富的recipes（即預制模板）來實現常見的分布式協調任務，包括共享鎖、互斥鎖、屏障、Leader選舉等。Curator的分布式鎖實現如InterProcessMutex和InterProcessSemaphoreMutex，直接提供了易于使用的API來獲取和釋放鎖。

Curator在實現分布式鎖時，充分考慮了ZooKeeper的特性，比如臨時節點的生命周期關聯會話、有序節點的排序機制以及Watcher事件的通知機制等，確保在各種異常情況下，鎖的行為符合預期，例如客戶端斷線后鎖能被正確釋放。

Curator內部集成了重試策略和背壓控制，當ZooKeeper操作遇到網絡延遲或短暫的ZooKeeper集群不穩定時，Curator能夠自動進行重試，而不是立即拋出異常。

@Component
public class ZkLock {

    private final CuratorFramework client;
    private final InterProcessMutex lock;

    @Value("${curator.zookeeper.connect-string}")
    private String zookeeperConnectString;

    public ZkLock() {
        RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 3);
        client = CuratorFrameworkFactory.newClient(zookeeperConnectString, retryPolicy);
        client.start();

        // 分布式鎖路徑
        String lockPath = "/locks/product_stock";
        lock = new InterProcessMutex(client, lockPath);
    }

    public void acquireLock(Runnable task) throws Exception {
        // 嘗試獲取鎖，超時時間為30秒
        if (lock.acquire(30, TimeUnit.SECONDS)) {
            try {
                task.run();  // 在持有鎖的情況下執行任務
            } finally {
                lock.release();  // 無論是否出現異常，都要確保釋放鎖
            }
        } else {
            throw new Exception("獲取分布是鎖失敗");
        }
    }
}

使用ZkLock：

@Override
public void confirmOrderWithZk(String customerNo, String orderNo) {
    // 查詢訂單信息
    OutboundOrderDO outboundOrderDO = outboundOrderMapper.selectOrderByOrderNo(customerNo, orderNo);
    String warehouseCode = outboundOrderDO.getWarehouseCode();

    // 查詢訂單明細  假設我們的出庫訂單是一單一件
    OutboundOrderDetailDO detailDO = orderDetailMapper.selectDetailByOrderNo(outboundOrderDO.getOrderNo());
    String sku = detailDO.getSku();
    Integer qty = detailDO.getQty();

    String lockKey = String.format("inventory:%s_%s_%s", customerNo, warehouseCode, sku);
    // 30秒過期
    zkLock.acquireLock(() -> {
        // 查詢庫存
        TbInventoryDO inventoryDO = tbInventoryMapper.selectSkuInventory(customerNo, warehouseCode, sku);
        Integer availableInventory = inventoryDO.getAvailableInventory();
        // 判斷庫存是否足夠
        if (qty > availableInventory){
            System.err.println("庫存不足，不能出庫");
            throw new ServiceException(StatusEnum.SERVICE_ERROR, "庫存不足，不能出庫");
        }

        // 扣減庫存
        TbInventoryDO updateInventory = new TbInventoryDO();
        updateInventory.setCustomerNo(customerNo);
        updateInventory.setWarehouseCode(warehouseCode);
        updateInventory.setSku(sku);
        // 庫存差值
        updateInventory.setDiffInventory(qty);
        tbInventoryMapper.updateInventory(updateInventory);
        
    });

}

Apache Curator實現的分布式鎖適用于需要在分布式環境中實現強一致性和高可靠性的并發控制場景，但是它對ZooKeeper的依賴就涉及到了一些網絡開銷以及運維復雜性等方面的缺點。

總結

其實現主要依賴分布式存儲系統或協調服務。常見的實現方式有如下幾種方式：

基于數據庫：利用數據庫事務的ACID特性，通過特定行的INSERT/UPDATE操作獲取鎖，DELETE/UPDATE操作釋放鎖。然而，可能存在性能瓶頸及高并發下數據庫連接數受限問題。
基于緩存系統（如Redis）：借助SETNX等原子操作或Lua腳本設置唯一鍵值對獲取鎖，并支持設置鎖超時以防止死鎖。這種方式具有較高的性能和內置防死鎖機制。
基于ZooKeeper：利用ZooKeeper的ZNode、觀察者機制及臨時有序節點。服務通過創建臨時節點競爭鎖，最小編號節點獲勝。節點故障時，ZooKeeper自動清理相關臨時節點，實現鎖的自動轉移。

而實際業務開發中，我們需要根據具體的業務以及系統資源等考慮，選擇合適的分布式鎖實現方式。

責任編輯：武曉燕來源：碼農Academy

分布式鎖系統