多線程並發問題解決之redis鎖

一 問題背景

        我們做的是醫療信息化系統，在系統中一條患者信息對醫院中當前科室中的所有診斷醫生是可見的，當有一個診斷醫生點擊按鈕處理該數據時，數據的狀態發生了變化，其他的醫生就不可以再處理此患者的數據了。我們開始的做法是，在醫生點擊按鈕時先去后台數據庫獲取當前數據狀態，根據狀態判斷數據是否可以操作，如果可以操作，則修改數據狀態，進行業務邏輯處理，否則提示數據已被其他人處理，不能處理。

二 問題分析

        按照上邊的業務邏輯，我們畫個圖分析，如下圖

在上圖中，如果用戶A和B同時向數據庫發起請求獲取數據狀態，數據庫返回wait，A和B都拿到了相同的狀態，判斷是可以操作數據的，這時他們處理數據。A用戶處理完成后提交了數據，數據庫狀態變為done，記錄此數據的處理人為A。由於B用戶也可以處理數據，所以他也提交數據，這時數據的操作人記錄為了B。有人會說，在A和B提交數據修改狀態時再做一個狀態的判斷，這種也難以避免最開始的獲取狀態的問題，即使這一步狀態獲取到了，提示后邊的人不能修改，這又會產生系統不友好的問題(我操作了半天，到最后你告訴我不能處理，我白忙活了)。以上問題產生的主要原因就是在多線程情況下對共享數據的資源競爭處理不當，我們需要保證數據的唯一性，即在某一時刻，只能有一個線程獨享數據資源。

三 問題解決

        如何解決呢？分布式鎖，分布式鎖有多種實現方式，本文我們用redis實現。由於redis是單線程的，所以一次只能處理一個請求，並將資源分配給這個請求，我們稱加 鎖。如下圖

多線程情況下，redis只會處理其中的一個，其他的暫時等待。如上圖當A和B同時發出請求時，redis接受並處理A請求，此時B請求排隊等待，等到A請求處理完后再處理B請求。此時redis已經將資源(lock)分配給了A，A請求數據庫，B請求沒有獲取到資源直接返回不在請求數據庫。這樣就保證了數據庫共享數據被唯一資源使用。代碼簡單實現

public class RedisLock {

    private static final String GET_RESULT = "OK";
    private static final String RELEASE_RESULT = "1";
    private static final String SET_IF_NOT_EXIST = "NX";
    private static final String SET_WITH_EXPIRE_TIME = "PX";

    /**
     * 獲取redis鎖
     * @param jedis redis客戶端
     * @param lockKey 鎖標識 key
     * @param requestId 鎖的持有者,加鎖的請求
     * @param expireTime 鎖過期時間
     * @return
     */
    public static boolean getLock(Jedis jedis, String lockKey, String requestId, int expireTime){
        //SET_IF_NOT_EXIST 當key不存在時 才處理
        //SET_WITH_EXPIRE_TIME 設置過期時間 時間由expireTime決定
        String result = jedis.set(lockKey, requestId, SET_IF_NOT_EXIST, SET_WITH_EXPIRE_TIME, expireTime);
        if (GET_RESULT.equals(result)) {
            return true;
        }
        return false;
    }

    /**
     * 釋放鎖
     * @param jedis
     * @param lockKey
     * @param requestId
     * @return
     */
    public static boolean releaseLock(Jedis jedis, String lockKey, String requestId){
        // 方式1
//        if (jedis.get(lockKey).equals(requestId)) {//校驗當前鎖的持有人與但概念請求是否相同
//            執行在這里時，如果鎖被其它請求重新獲取到了，此時就不該刪除了
//            jedis.del(lockKey);
//        }

        //方式2
        // eval() 方法會交給redis服務端執行，減少了從服務端再到客戶端處理的過程
        //賦值 KEYS[1] = lockKey   ARGV[1] = requestId
        String script = "if redis.call('get', KEYS[1]) == ARGV[1] then return redis.call('del', KEYS[1]) else return 0 end";
        Object releaseResult = jedis.eval(script, Collections.singletonList(lockKey), Collections.singletonList(requestId));
        if (RELEASE_RESULT.equals(releaseResult.toString())) {
            return true;
        }
        return false;
    }
}

四 測試鎖機制

　　測試並發我們可以使用一些軟件，比如Jmeter，本文我們寫個方法測試

public static void main(String[] args) {
        //要創建的線程的數量
        CountDownLatch looker = new CountDownLatch(1);
        CountDownLatch latch = new CountDownLatch(10);
        final String key = "lockKey";
        for(int i=0; i < latch.getCount(); i++){
            Jedis jedis = new Jedis();
            UUID uuid = UUID.randomUUID();
            Thread thread = new Thread(new Runnable() {
                @Override
                public void run() {
                    try {
                        looker.await();
                        System.out.println(Thread.currentThread().getName()+"競爭資源，獲取鎖");
                        boolean getResult = getLock(jedis, key, uuid.toString(), 5000);
                        if(getResult){
                            System.out.println(Thread.currentThread().getName()+"獲取到了鎖，處理業務，用時3秒");
                            Thread.sleep(3000);
                            boolean releaseResult = releaseLock(jedis, key, uuid.toString());
                            if(releaseResult){
                                System.out.println(Thread.currentThread().getName()+"業務處理完畢，釋放鎖");
                            }
                        }else{
                            System.out.println(Thread.currentThread().getName()+"競爭資源失敗，未獲取到鎖");
                        }
                        latch.countDown();
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            });
            thread.start();
        }

        try {
            System.out.println("准備，5秒后開始");
            Thread.sleep(5000);
            looker.countDown(); //發令  let all threads proceed

            latch.await(); // // wait for all to finish
            System.out.println("結束");
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

    }

可以看到控制台上輸出的結果