* @date 2018/5/10 12:35 * @date 2018/8/28 9:27
*/
@RestController
@RequestMapping("/redis")
public class RedisController {
private static LongAdder longAdder = new LongAdder();
private static Long LOCK_EXPIRE_TIME = 200L;
private static Long stock = 10000L;
@Autowired
private RedisService redisService;
static {
longAdder.add(10000L);
}
@GetMapping("/v1/seckill")
public String seckillV1() {
Long time = System.currentTimeMillis() + LOCK_EXPIRE_TIME;
if (!redisService.lock(SeckillKeyPrefix.seckillKeyPrefix, "redis-seckill", String.valueOf(time))) {
return "人太多了，換個姿勢操作一下";
}
if (longAdder.longValue() == 0L) {
return "已搶光";
}
doSomeThing();
if (longAdder.longValue() == 0L) {
return "已搶光";
}
longAdder.decrement();
redisService.unlock(SeckillKeyPrefix.seckillKeyPrefix, "redis-seckill", String.valueOf(time));
Long stock = longAdder.longValue();
Long bought = 10000L - stock;
return "已搶" + bought + ", 還剩下" + stock;
}
@GetMapping("/detail")
public String detail() {
Long stock = longAdder.longValue();
Long bought = 10000L - stock;
return "已搶" + bought + ", 還剩下" + stock;
}
@GetMapping("/v2/seckill")
public String seckillV2() {
if (longAdder.longValue() == 0L) {
return "已搶光";
}
doSomeThing();
if (longAdder.longValue() == 0L) {
return "已搶光";
}
longAdder.decrement();
Long stock = longAdder.longValue();
Long bought = 10000L - stock;
return "已搶" + bought + ", 還剩下" + stock;
}
@GetMapping("/v3/seckill")
public String seckillV3() {
if (stock == 0) {
return "已搶光";
}
doSomeThing();
stock--;
Long bought = 10000L - stock;
return "已搶" + bought + ", 還剩下" + stock;
}
public void doSomeThing() {
try {
TimeUnit.MILLISECONDS.sleep(100);
} catch (InterruptedException ex) {
ex.printStackTrace();
}
}
}
複製代碼
對http://localhost:8081/redis/v1/seckill進行壓測，我使用的壓測工具是ab測試工具。這裡用10000個併發用戶，20000個請求來進行壓測。
ab -c 10000 -n 20000 http://localhost:8081/redis/v1/seckill
複製代碼
壓測結果如下：
E:\cmazxiaoma_download\httpd-2.4.34-o102o-x64-vc14\Apache24\bin>ab -c 10000 -n 2
0000 http://localhost:8081/redis/v1/seckill
This is ApacheBench, Version 2.3
Copyright 1996 Adam Twiss, Zeus Technology Ltd, http://www.zeustech.net/
Licensed to The Apache Software Foundation, http://www.apache.org/
Benchmarking localhost (be patient)
Completed 2000 requests
Completed 4000 requests
Completed 6000 requests
Completed 8000 requests
Completed 10000 requests
Completed 12000 requests
Completed 14000 requests
Completed 16000 requests
Completed 18000 requests
Completed 20000 requests
Finished 20000 requests
Server Software:
Server Hostname: localhost
Server Port: 8081
Document Path: /redis/v1/seckill
Document Length: 22 bytes
Concurrency Level: 10000
Time taken for tests: 108.426 seconds
Complete requests: 20000
Failed requests: 19991
(Connect: 0, Receive: 0, Length: 19991, Exceptions: 0)
Total transferred: 3420218 bytes
HTML transferred: 760218 bytes
Requests per second: 184.46 [#/sec] (mean)
Time per request: 54213.000 [ms] (mean)
Time per request: 5.421 [ms] (mean, across all concurrent requests)
Transfer rate: 30.80 [Kbytes/sec] received
Connection Times (ms)
min mean[+/-sd] median max
Connect: 0 0 6.3 0 549
Processing: 2393 36477 16329.1 45101 90269
Waiting: 182 36435 16351.4 45046 90267
Total: 2393 36477 16329.0 45101 90269
Percentage of the requests served within a certain time (ms)
50% 45101
66% 47680
75% 49136
80% 50392
90% 53200
95% 53743
98% 54510
99% 56014
100% 90269 (longest request)
複製代碼
我們再來看看是否有超賣現象，貌似還是正常。
回溯分析
我打開RedisDesktopManager查看db0的key信息時，發現還有一個key沒有刪除掉。說明我們寫的unlock()方法在1w併發用戶，2w請求下還是存在問題。
仔細推敲自己之前寫的代碼發現(還是拿上面的例子說事)，客戶端D雖然獲取鎖失敗，但是之前進行了String oldValue = jedis.getSet(realKey, value);操作，還是成功的更新了realKey對應的value。我們進行unlock()操作時，釋放客戶端的鎖是根據value來標識當前客戶端的。一開始客戶端C的value是T2，由於客戶端D的getSet()操作，覆蓋掉了客戶端C的value，讓其更新成T3。由於value.equals(currentValue)條件不成立，所以不會執行到jedis.del(realKey)
其實lock()方法也經不起推敲: 1.分佈式各個系統時間不一致，如果要這樣做，只能進行時間同步。 2.當某個客戶端鎖過期時，多個客戶端開始爭搶鎖。雖然最後只有一個客戶端能成功鎖，但是獲取鎖失敗的客戶端能覆蓋獲取鎖成功客戶端的過期時間。 3.當客戶端的鎖過期時間被覆蓋，會造成鎖不具有標識性，會造成客戶端沒有釋放鎖。
所以我們要重寫lock與unlock()的邏輯，看到網上已經有很多的解決方案。(不過也有很多錯誤案例)
我們可以通過redis的set(key,value,NX,EX,timeout)合併普通的set()和expire()操作，使其具有原子性。
/**
* Set the string value as value of the key. The string can't be longer than 1073741824 bytes (1
* GB).
* @param key
* @param value
* @param nxxx NX|XX, NX -- Only set the key if it does not already exist. XX -- Only set the key
* if it already exist.
* @param expx EX|PX, expire time units: EX = seconds; PX = milliseconds
* @param time expire time in the units ofexpx
* @return Status code reply
*/
public String set(final String key, final String value, final String nxxx, final String expx,
final long time) {
checkIsInMultiOrPipeline();
client.set(key, value, nxxx, expx, time);
return client.getStatusCodeReply();
}
複製代碼
通過set(key,value,NX,EX,timeout)方法，我們就可以輕鬆實現分佈式鎖。值得注意的是這裡的value作為客戶端鎖的唯一標識，不能重複。
public boolean lock1(KeyPrefix prefix, String key, String value, Long lockExpireTimeOut,
Long lockWaitTimeOut) {
Jedis jedis = null;
try {
jedis = jedisPool.getResource();
String realKey = prefix.getPrefix() + key;
Long deadTimeLine = System.currentTimeMillis() + lockWaitTimeOut;
for (;;) {
String result = jedis.set(realKey, value, "NX", "PX", lockExpireTimeOut);
if ("OK".equals(result)) {
return true;
}
lockWaitTimeOut = deadTimeLine - System.currentTimeMillis();
if (lockWaitTimeOut <= 0L) {
return false;
}
}
} catch (Exception ex) {
log.info("lock error");
} finally {
returnToPool(jedis);
}
return false;
}
複製代碼
我們可以使用lua腳本合併get()和del()操作，使其具有原子性。一切大功告成。
public boolean unlock1(KeyPrefix prefix, String key, String value) {
Jedis jedis = null;
try {
jedis = jedisPool.getResource();
String realKey = prefix.getPrefix() + key;
String luaScript = "if redis.call('get', KEYS[1]) == ARGV[1] then return redis.call('del', KEYS[1]) else return 0 end";
Object result = jedis.eval(luaScript, Collections.singletonList(realKey),
Collections.singletonList(value));
if ("1".equals(result)) {
return true;
}
} catch (Exception ex) {
log.info("unlock error");
} finally {
returnToPool(jedis);
}
return false;
}
複製代碼
剛才看了評論，看到了各位大佬提出的一系列問題。我做出以下解釋：
- 秒殺操作，我在這裡只是舉一個例子，更直觀的判斷分佈式鎖可行,不適合實際場景！！！！！實際上搶購，是將商品庫存放入到redis、將是否結束標記Flag放入到內存中，通過內存標記和redis中的decr()預減庫存，然後將秒殺消息入隊到消息隊列中，最後消費消息並落地到DB中。
2.請耐心讀完本篇文章。第一個案例代碼是錯誤的，我後續講解了如何發現和分析錯誤案例代碼的思路。 在此基礎下，推導出正確的代碼。
3.通過評論，我看到有一篇文章作者的思路是這樣的： 獲取鎖之後，通過標誌位和開啟新線程的方式輪詢去刷新當前客戶端持有鎖的時間，以保證在釋放鎖之前鎖不會過期，然後鎖釋放後，將標誌位置為false，線程停止循環。但是這樣有一個問題：假如執行了lock()操作之後，客戶端由於一些原因阻塞了，那麼unlock()方法一直得不到執行，那麼標誌位一直為true，開啟刷新過期時間的線程一直死循環，會造成資源的嚴重浪費。而且線程一直增加當前客戶端持有鎖的時間，會造成其他客戶端一直拿不到鎖，而且造成死鎖。
喜歡的點點關注，私信回覆‘資料’有驚喜
閱讀更多 JavaSpring高級進階 的文章
