使用Redis實現分散式鎖

關於分散式鎖的實現，我的前一篇文章講解了如何使用Zookeeper實現分散式鎖。關於分散式鎖的背景此處不再做贅述，我們直接討論下如何使用Redis實現分散式鎖。

關於Redis，筆主不打算做長篇大論的介紹，只介紹下Redis優秀的特性。

1. 支援豐富的資料型別，如String、List、Map、Set、ZSet等。
2. 支援資料持久化，RDB和AOF兩種方式
3. 支援叢集工作模式，分割槽容錯性強
4. 單執行緒，順序處理命令
5. 支援事務
6. 支援釋出與訂閱

Redis實現分散式鎖使用了SETNX命令，我們可以參考Redis官方對此命令的介紹。

SETNX key value 將 key 的值設為 value ，當且僅當 key 不存在。 若給定的 key 已經存在，則 SETNX 不做任何動作。 SETNX 是set if not exists的簡寫。 官方介紹地址：

http://doc.redisfans.com/string/setnx.html

當然，Redis實現分散式鎖也利用了單執行緒順序處理命令的特性。程式碼實現也是比較簡單的，筆主使用了SpringBoot進行開發，SpringBoot提供了操作Redis的工具類RedisTemplate。

首先，我們需要封裝一個公共的Redis訪問工具類。該類需要注入RedisTemplate例項和ValueOperations例項，使用ValueOperations例項是因為Redis實現的分散式鎖使用了最簡單的String型別。另外，我們需要封裝3個方法，分別是setIfObsent (String key, String value)、 expire (String key, long timeout, TimeUnit unit) 、delete (String key) ，分別對應Redis的SETNX、expire、del命令。以下是Redis訪問工具類的具體實現：

/**
 * Redis訪問工具類
 * @author zhaoheng
 * @date   2018年8月10日
 */
@Component
public class RedisDao {

	@Autowired
	private RedisTemplate redisTemplate;
	
	@Resource(name="redisTemplate")
	private ValueOperations<Object, Object> valOpsObj;
	
	/**
	 * 如果key不存在，就儲存一個key-value，相當於SETNX命令
	 * @param key      鍵
	 * @param value    值，可以為空
	 * @return
	 */
	public boolean setIfObsent (String key, String value) {
		return valOpsObj.setIfAbsent(key, value);
	}
	
	/**
	 * 為key設定失效時間
	 * @param key       鍵
	 * @param timeout   時間大小
	 * @param unit      時間單位
	 */
	public boolean expire (String key, long timeout, TimeUnit unit) {
		return redisTemplate.expire(key, timeout, unit);
	}
	
	/**
	 * 刪除key
	 * @param key 鍵
	 */
	public void delete (String key) {
		redisTemplate.delete(key);
	}
}
 

完成了Redis訪問工具類的實現，現在需要考慮的是如何去模擬競爭分散式鎖。因為Redis本身就是支援分散式叢集的，所以只需要模擬出多執行緒處理業務場景。這裡筆主採用執行緒池來模擬，以下是測試類的具體實現：

@RestController
@RequestMapping("test")
public class TestController {

	private static final Logger LOG = LoggerFactory.getLogger(TestController.class);  //日誌物件
	@Autowired
	private RedisDao redisDao;
	//定義的分散式鎖key
	private static final String LOCK_KEY = "MyTestLock";
	
	@RequestMapping(value={"testRedisLock"}, method=RequestMethod.GET)
	public void testRedisLock () {
		ExecutorService executorService = Executors.newFixedThreadPool(5);
		for (int i = 0; i < 5; i++) {
			executorService.submit(new Runnable() {
				@Override
				public void run() {
				    //獲取分散式鎖
					boolean flag = redisDao.setIfObsent(LOCK_KEY, "lock");
					if (flag) {
						LOG.info(Thread.currentThread().getName() + "：獲取Redis分散式鎖成功");
						//獲取鎖成功後設置失效時間
						redisDao.expire(LOCK_KEY, 2, TimeUnit.SECONDS);
						try {
							LOG.info(Thread.currentThread().getName() + "：處理業務開始");
							Thread.sleep(1000); //睡眠1000ms模擬處理業務
							LOG.info(Thread.currentThread().getName() + "：處理業務結束");
							//處理業務完成後刪除鎖
							redisDao.delete(LOCK_KEY);
						} catch (InterruptedException e) {
							LOG.error("處理業務異常：", e);
						}
					} else {
						LOG.info(Thread.currentThread().getName() + "：獲取Redis分散式鎖失敗");
					}
				}
			});
		}
	}
}

通過上面這段程式碼，可能會產生以下幾個疑問：

1. 執行緒如果獲取分散式鎖失敗，為什麼不嘗試重新獲取鎖?
2. 執行緒獲取分散式鎖成功後，設定了鎖的失效時間，這個失效時間長短如何確定？
3. 執行緒業務處理結束後，為什麼要做刪除鎖的操作？

針對這幾個疑問，我們可以來討論下。

1. 第一，Redis的SETNX命令，如果key已經存在，則不會做任何操作，所以SETNX實現的分散式鎖並不是可重入鎖。當然，也可以自己通過程式碼實現重試n次或者直至獲取到分散式鎖為止。但是，這不能保證競爭的公平性，某個執行緒會因為一直等待鎖而阻塞。因此，Redis實現的分散式鎖更適用於對共享資源一寫多讀的場景。
2. 第二，分散式鎖必須設定失效時間，而且失效時間必須大於業務處理所需的時間（保證資料一致性）。所以，在測試階段儘可能準確的預測出業務正常處理所需的時間，設定失效時間是防止因為業務處理過程的某些原因導致死鎖的情況。
3. 第三，業務處理結束，必須要做刪除鎖的操作。