Redis實現庫存扣減的解決方案防止商品超賣
更新時間：2022年06月28日 15:57:19 作者：Resourceful!
在日常開發(fā)中有很多地方都有類似扣減庫存的操作，比如電商系統(tǒng)中的商品庫存，抽獎系統(tǒng)中的獎品庫存等，基于redis實現扣減庫存的具體實現，初始化庫存回調函數（IStockCallback）扣減庫存服務（StockService），感興趣的朋友跟隨小編一起看看吧
Redis 如何實現庫存扣減操作？如何防止商品被超賣？
基于數據庫單庫存 基于數據庫多庫存 基于redis 基于redis實現扣減庫存的具體實現 初始化庫存回調函數（IStockCallback） 扣減庫存服務（StockService）。
在日常開發(fā)中有很多地方都有類似扣減庫存的操作，比如電商系統(tǒng)中的商品庫存，抽獎系統(tǒng)中的獎品庫存等。
解決方案
1. 使用mysql數據庫
使用一個字段來存儲庫存，每次扣減庫存去更新這個字段。
2. 還是使用數據庫
但是將庫存分層多份存到多條記錄里面，扣減庫存的時候路由一下，這樣子增大了并發(fā)量，但是還是避免不了大量的去訪問數據庫來更新庫存。
3. 將庫存放到redis使用redis的incrby特性來扣減庫存。
分析
在上面的第一種和第二種方式都是基于數據來扣減庫存。
[基于數據庫單庫存]
第一種方式在所有請求都會在這里等待鎖，獲取鎖有去扣減庫存。在并發(fā)量不高的情況下可以使用，但是一旦并發(fā)量大了就會有大量請求阻塞在這里，導致請求超時，進而整個系統(tǒng)雪崩；而且會頻繁的去訪問數據庫，大量占用數據庫資源，所以在并發(fā)高的情況下這種方式不適用。
[基于數據庫多庫存]
第二種方式其實是第一種方式的優(yōu)化版本，在一定程度上提高了并發(fā)量，但是在還是會大量的對數據庫做更新操作大量占用數據庫資源。
基于數據庫來實現扣減庫存還存在的一些問題：
用數據庫扣減庫存的方式，扣減庫存的操作必須在一條語句中執(zhí)行，不能先selec在update，這樣在并發(fā)下會出現超扣的情況。如：
update number set x=x-1 where x > 0
- MySQL自身對于高并發(fā)的處理性能就會出現問題，一般來說，MySQL的處理性能會隨著并發(fā)thread上升而上升，但是到了一定的并發(fā)度之后會出現明顯的拐點，之后一路下降，最終甚至會比單thread的性能還要差。
- 當減庫存和高并發(fā)碰到一起的時候，由于操作的庫存數目在同一行，就會出現爭搶InnoDB行鎖的問題，導致出現互相等待甚至死鎖，從而大大降低MySQL的處理性能，最終導致前端頁面出現超時異常。
[基于redis]
針對上述問題的問題我們就有了第三種方案，將庫存放到緩存，利用redis的incrby特性來扣減庫存，解決了超扣和性能問題。但是一旦緩存丟失需要考慮恢復方案。比如抽獎系統(tǒng)扣獎品庫存的時候，初始庫存=總的庫存數-已經發(fā)放的獎勵數，但是如果是異步發(fā)獎，需要等到MQ消息消費完了才能重啟redis初始化庫存，否則也存在庫存不一致的問題。
基于redis實現扣減庫存的具體實現
- 我們使用redis的lua腳本來實現扣減庫存
- 由于是分布式環(huán)境下所以還需要一個分布式鎖來控制只能有一個服務去初始化庫存
- 需要提供一個回調函數，在初始化庫存的時候去調用這個函數獲取初始化庫存
[初始化庫存回調函數(IStockCallback )]
/**
* 獲取庫存回調
* @author yuhao.wang
*/
public interface IStockCallback {
/**
* 獲取庫存
* @return
*/
int getStock();
}[扣減庫存服務（StockService）]
/**
* 扣庫存
*
* @author yuhao.wang
*/
@Service
public class StockService {
Logger logger = LoggerFactory.getLogger(StockService.class);
/**
* 不限庫存
*/
public static final long UNINITIALIZED_STOCK = -3L;
/**
* Redis 客戶端
*/
@Autowired
private RedisTemplate<String, Object> redisTemplate;
/**
* 執(zhí)行扣庫存的腳本
*/
public static final String STOCK_LUA;
static {
/**
*
* @desc 扣減庫存Lua腳本
* 庫存（stock）-1：表示不限庫存
* 庫存（stock）0：表示沒有庫存
* 庫存（stock）大于0：表示剩余庫存
*
* @params 庫存key
* @return
* -3:庫存未初始化
* -2:庫存不足
* -1:不限庫存
* 大于等于0:剩余庫存（扣減之后剩余的庫存）
* redis緩存的庫存(value)是-1表示不限庫存，直接返回1
*/
StringBuilder sb = new StringBuilder();
sb.append("if (redis.call('exists', KEYS[1]) == 1) then");
sb.append(" local stock = tonumber(redis.call('get', KEYS[1]));");
sb.append(" local num = tonumber(ARGV[1]);");
sb.append(" if (stock == -1) then");
sb.append(" return -1;");
sb.append(" end;");
sb.append(" if (stock >= num) then");
sb.append(" return redis.call('incrby', KEYS[1], 0 - num);");
sb.append(" end;");
sb.append(" return -2;");
sb.append("end;");
sb.append("return -3;");
STOCK_LUA = sb.toString();
}
/**
* @param key 庫存key
* @param expire 庫存有效時間,單位秒
* @param num 扣減數量
* @param stockCallback 初始化庫存回調函數
* @return -2:庫存不足; -1:不限庫存; 大于等于0:扣減庫存之后的剩余庫存
*/
public long stock(String key, long expire, int num, IStockCallback stockCallback) {
long stock = stock(key, num);
// 初始化庫存
if (stock == UNINITIALIZED_STOCK) {
RedisLock redisLock = new RedisLock(redisTemplate, key);
try {
// 獲取鎖
if (redisLock.tryLock()) {
// 雙重驗證，避免并發(fā)時重復回源到數據庫
stock = stock(key, num);
if (stock == UNINITIALIZED_STOCK) {
// 獲取初始化庫存
final int initStock = stockCallback.getStock();
// 將庫存設置到redis
redisTemplate.opsForValue().set(key, initStock, expire, TimeUnit.SECONDS);
// 調一次扣庫存的操作
stock = stock(key, num);
}
}
} catch (Exception e) {
logger.error(e.getMessage(), e);
} finally {
redisLock.unlock();
}
}
return stock;
}
/**
* 加庫存(還原庫存)
*
* @param key 庫存key
* @param num 庫存數量
* @return
*/
public long addStock(String key, int num) {
return addStock(key, null, num);
}
/**
* 加庫存
*
* @param key 庫存key
* @param expire 過期時間（秒）
* @param num 庫存數量
* @return
*/
public long addStock(String key, Long expire, int num) {
boolean hasKey = redisTemplate.hasKey(key);
// 判斷key是否存在，存在就直接更新
if (hasKey) {
return redisTemplate.opsForValue().increment(key, num);
}
Assert.notNull(expire,"初始化庫存失敗，庫存過期時間不能為null");
RedisLock redisLock = new RedisLock(redisTemplate, key);
try {
if (redisLock.tryLock()) {
// 獲取到鎖后再次判斷一下是否有key
hasKey = redisTemplate.hasKey(key);
if (!hasKey) {
// 初始化庫存
redisTemplate.opsForValue().set(key, num, expire, TimeUnit.SECONDS);
}
}
} catch (Exception e) {
logger.error(e.getMessage(), e);
} finally {
redisLock.unlock();
}
return num;
}
/**
* 獲取庫存
*
* @param key 庫存key
* @return -1:不限庫存; 大于等于0:剩余庫存
*/
public int getStock(String key) {
Integer stock = (Integer) redisTemplate.opsForValue().get(key);
return stock == null ? -1 : stock;
}
/**
* 扣庫存
*
* @param key 庫存key
* @param num 扣減庫存數量
* @return 扣減之后剩余的庫存【-3:庫存未初始化; -2:庫存不足; -1:不限庫存; 大于等于0:扣減庫存之后的剩余庫存】
*/
private Long stock(String key, int num) {
// 腳本里的KEYS參數
List<String> keys = new ArrayList<>();
keys.add(key);
// 腳本里的ARGV參數
List<String> args = new ArrayList<>();
args.add(Integer.toString(num));
long result = redisTemplate.execute(new RedisCallback<Long>() {
@Override
public Long doInRedis(RedisConnection connection) throws DataAccessException {
Object nativeConnection = connection.getNativeConnection();
// 集群模式和單機模式雖然執(zhí)行腳本的方法一樣，但是沒有共同的接口，所以只能分開執(zhí)行
// 集群模式
if (nativeConnection instanceof JedisCluster) {
return (Long) ((JedisCluster) nativeConnection).eval(STOCK_LUA, keys, args);
}
// 單機模式
else if (nativeConnection instanceof Jedis) {
return (Long) ((Jedis) nativeConnection).eval(STOCK_LUA, keys, args);
}
return UNINITIALIZED_STOCK;
}
});
return result;
}
}[調用]
/**
* @author yuhao.wang
*/
@RestController
public class StockController {
@Autowired
private StockService stockService;
@RequestMapping(value = "stock", produces = MediaType.APPLICATION_JSON_UTF8_VALUE)
public Object stock() {
// 商品ID
long commodityId = 1;
// 庫存ID
String redisKey = "redis_key:stock:" + commodityId;
long stock = stockService.stock(redisKey, 60 * 60, 2, () -> initStock(commodityId));
return stock >= 0;
}
/**
* 獲取初始的庫存
*
* @return
*/
private int initStock(long commodityId) {
// TODO 這里做一些初始化庫存的操作
return 1000;
}
@RequestMapping(value = "getStock", produces = MediaType.APPLICATION_JSON_UTF8_VALUE)
public Object getStock() {
// 商品ID
long commodityId = 1;
// 庫存ID
String redisKey = "redis_key:stock:" + commodityId;
return stockService.getStock(redisKey);
}
@RequestMapping(value = "addStock", produces = MediaType.APPLICATION_JSON_UTF8_VALUE)
public Object addStock() {
// 商品ID
long commodityId = 2;
// 庫存ID
String redisKey = "redis_key:stock:" + commodityId;
return stockService.addStock(redisKey, 2);
}
}
結語
到此這篇關于Redis 如何實現庫存扣減操作？如何防止商品被超賣？的文章就介紹到這了,更多相關Redis庫存扣減內容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關文章希望大家以后多多支持腳本之家！
相關文章
redis-cli 是Redis命令行界面,一個簡單的程序,允許直接從終端向Redis發(fā)送命令,并讀取服務器發(fā)送的回復,本文主要介紹了redis中redis-cli使用小結,感興趣的可以了解一下2023-10-10
生成全局唯一的標識符是非常常見的需求,本文主要介紹了redis生成全局id的實現步驟,文中通過示例代碼介紹的非常詳細,對大家的學習或者工作具有一定的參考學習價值,需要的朋友們下面隨著小編來一起學習學習吧2024-05-05
為了提高數據處理效率和降低存儲成本,優(yōu)化數據結構和采用高效的存儲策略至關重要,使用最小存儲形式、整數編碼、Redis的HyperLogLog等方法可以有效減少內存占用,Redis6引入的對象壓縮、設置合理的過期時間、數據分片2024-09-09
這篇文章主要介紹了Redis?布隆過濾器的相關命令的使用,本文給大家介紹的非常詳細，對大家的學習或工作具有一定的參考借鑒價值，需要的朋友可以參考下2022-08-08
本文主要介紹了Redis數據遷移RedisShake的實現方法,文中通過示例代碼介紹的非常詳細,具有一定的參考價值,感興趣的小伙伴們可以參考一下2022-04-04
事務是關系型數據庫的特征之一，那么作為 Nosql 的代表 Redis 中有事務嗎？如果有，那么 Redis 當中的事務又是否具備關系型數據庫的 ACID 四大特性，本文就來詳細介紹一下2021-08-08
這篇文章主要為大家詳細介紹了項目開發(fā)時如果出現線上Redis一直報連接超時的問題該如何解決，文中的示例代碼簡潔易懂，需要的小伙伴可以借鑒一下2023-08-08
Redis是內存數據庫，數據都是存儲在內存中，需要定期將Redis中的數據以某種形式(或命數據令)從內存保存到硬盤，今天給大家分享Redis數據的持久化的概念介紹，需要的朋友參考下吧2021-08-08
Redis使用HyperLogLog的主要作用是在大數據流（view，IP，城市）的情況下進行去重計數，這篇文章主要介紹了Redis中?HyperLogLog數據類型使用總結,需要的朋友可以參考下2023-03-03
今天為大家介紹下linux系統(tǒng)下redis的主從配置方法,Linux系統(tǒng)下的redis的主從配置方法非常簡單下面是具體的操作步驟2018-09-09