当前位置：网站首页>基于redis实现的扣减库存
基于redis实现的扣减库存
2022-06-29 03:26:00 【JackieZhengChina】
使用mysql数据库，使用一个字段来存储库存，每次扣减库存去更新这个字段。
还是使用数据库，但是将库存分层多份存到多条记录里面，扣减库存的时候路由一下，这样子增大了并发量，但是还是避免不了大量的去访问数据库来更新库存。
将库存放到redis使用redis的incrby特性来扣减库存。
用数据库扣减库存的方式，扣减库存的操作必须在一条语句中执行，不能先selec在update，这样在并发下会出现超扣的情况。如：
MySQL自身对于高并发的处理性能就会出现问题，一般来说，MySQL的处理性能会随着并发thread上升而上升，但是到了一定的并发度之后会出现明显的拐点，之后一路下降，最终甚至会比单thread的性能还要差。
当减库存和高并发碰到一起的时候，由于操作的库存数目在同一行，就会出现争抢InnoDB行锁的问题，导致出现互相等待甚至死锁，从而大大降低MySQL的处理性能，最终导致前端页面出现超时异常。
我们使用redis的lua脚本来实现扣减库存
由于是分布式环境下所以还需要一个分布式锁来控制只能有一个服务去初始化库存
需要提供一个回调函数，在初始化库存的时候去调用这个函数获取初始化库存
/**
* 获取库存回调
* @author yuhao.wang
*/
public interface IStockCallback {
/**
* 获取库存
* @return
*/
int getStock();
}
package com.xiaolyuh.service;
import com.xiaolyuh.lock.RedisLock;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.dao.DataAccessException;
import org.springframework.data.redis.connection.RedisConnection;
import org.springframework.data.redis.core.RedisCallback;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Service;
import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisCluster;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.TimeUnit;
/**
* 扣库存
*
* @author yuhao.wang
*/
@Service
public class StockService {
Logger logger = LoggerFactory.getLogger(StockService.class);
/**
* 库存不足
*/
public static final int LOW_STOCK = 0;
/**
* 不限库存
*/
public static final long UNINITIALIZED_STOCK = -1L;
/**
* Redis 客户端
*/
@Autowired
private RedisTemplate<String, Object> redisTemplate;
/**
* 执行扣库存的脚本
*/
public static final String STOCK_LUA;
static {
/**
*
* @desc 扣减库存Lua脚本
* 库存（stock）-1：表示不限库存
* 库存（stock）0：表示没有库存
* 库存（stock）大于0：表示剩余库存
*
* @params 库存key
* @return
* 0:库存不足
* -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(" if (stock == -1) then");
sb.append(" return 1;");
sb.append(" end;");
sb.append(" if (stock > 0) then");
sb.append(" redis.call('incrby', KEYS[1], -1);");
sb.append(" return stock;");
sb.append(" end;");
sb.append(" return 0;");
sb.append("end;");
sb.append("return -1;");
STOCK_LUA = sb.toString();
}
/**
* @param key 库存key
* @param expire 库存有效时间,单位秒
* @param stockCallback 初始化库存回调函数
* @return 0:库存不足; -1:库存未初始化; 大于0:扣减库存之前的剩余库存（扣减之前剩余的库存）
*/
public long stock(String key, long expire, IStockCallback stockCallback) {
long stock = stock(key);
// 初始化库存
if (stock == UNINITIALIZED_STOCK) {
RedisLock redisLock = new RedisLock(redisTemplate, key);
try {
// 获取锁
if (redisLock.tryLock()) {
// 双重验证，避免并发时重复回源到数据库
stock = stock(key);
if (stock == UNINITIALIZED_STOCK) {
// 获取初始化库存
final int initStock = stockCallback.getStock();
// 将库存设置到redis
redisTemplate.opsForValue().set(key, initStock, expire, TimeUnit.SECONDS);
// 调一次扣库存的操作
stock = stock(key);
}
}
} catch (Exception e) {
logger.error(e.getMessage(), e);
} finally {
redisLock.unlock();
}
}
return stock;
}
/**
* 获取库存
*
* @param key 库存key
* @return 0:库存不足; -1:库存未初始化; 大于0:剩余库存
*/
public int getStock(String key) {
Integer stock = (Integer) redisTemplate.opsForValue().get(key);
return stock == null ? -1 : stock;
}
/**
* 扣库存
*
* @param key 库存key
* @return 扣减之前剩余的库存【0:库存不足; -1:库存未初始化; 大于0:扣减库存之前的剩余库存】
*/
private Long stock(String key) {
// 脚本里的KEYS参数
List<String> keys = new ArrayList<>();
keys.add(key);
// 脚本里的ARGV参数
List<String> args = new ArrayList<>();
long result = redisTemplate.execute(new RedisCallback<Long>() {
@Override
public Long doInRedis(RedisConnection connection) throws DataAccessException {
Object nativeConnection = connection.getNativeConnection();
// 集群模式和单机模式虽然执行脚本的方法一样，但是没有共同的接口，所以只能分开执行
// 集群模式
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, () -> 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);
}
}
理论上即使是这样由于MySQL事务的特性，这种方法只能降低超卖的数量，但是不可能完全避免超扣。因为数据库默认隔离级别是repeatable read，假如库存是5，有A、B两个请求分别创建了事务并且都没有提交，当A事务提交了，改了库存为4，但是因为是事务隔离级别是可重复读的，所有B看不到A事务改的库存。到时B看到的库存还是5，所以B修改库存为4，这样就出现了超扣问题。所以我们扣库存的时候需要将事务隔离级别设置成read commit才可以。（我自己测试没有出现这种情况）
MySQL自身对于高并发的处理性能就会出现问题，一般来说，MySQL的处理性能会随着并发thread上升而上升，但是到了一定的并发度之后会出现明显的拐点，之后一路下降，最终甚至会比单thread的性能还要差。
当减库存和高并发碰到一起的时候，由于操作的库存数目在同一行，就会出现争抢InnoDB行锁的问题，导致出现互相等待甚至死锁，从而大大降低MySQL的处理性能，最终导致前端页面出现超时异常。
基于redis
针对上述问题的问题我们就有了第三种方案，将库存放到缓存，利用redis的incrby特性来扣减库存，解决了超扣和性能问题。但是一旦缓存丢失需要考虑恢复方案。比如抽奖系统扣奖品库存的时候，初始库存=总的库存数-已经发放的奖励数，但是如果是异步发奖，需要等到MQ消息消费完了才能重启redis初始化库存，否则也存在库存不一致的问题。
基于redis实现扣减库存的具体实现
- 我们使用redis的lua脚本来实现扣减库存
- 由于是分布式环境下所以还需要一个分布式锁来控制只能有一个服务去初始化库存
- 需要提供一个回调函数，在初始化库存的时候去调用这个函数获取初始化库存
初始化库存回调函数(IStockCallback )
update number set x=x-1 where x > 0
扣减库存服务（StockService）
/**
* 获取库存回调
* @author yuhao.wang
*/
public interface IStockCallback {
/**
* 获取库存
* @return
*/
int getStock();
}
调用
package com.xiaolyuh.service;
import com.xiaolyuh.lock.RedisLock;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.dao.DataAccessException;
import org.springframework.data.redis.connection.RedisConnection;
import org.springframework.data.redis.core.RedisCallback;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Service;
import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisCluster;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.TimeUnit;
/**
* 扣库存
*
* @author yuhao.wang
*/
@Service
public class StockService {
Logger logger = LoggerFactory.getLogger(StockService.class);
/**
* 库存不足
*/
public static final int LOW_STOCK = 0;
/**
* 不限库存
*/
public static final long UNINITIALIZED_STOCK = -1L;
/**
* Redis 客户端
*/
@Autowired
private RedisTemplate<String, Object> redisTemplate;
/**
* 执行扣库存的脚本
*/
public static final String STOCK_LUA;
static {
/**
*
* @desc 扣减库存Lua脚本
* 库存（stock）-1：表示不限库存
* 库存（stock）0：表示没有库存
* 库存（stock）大于0：表示剩余库存
*
* @params 库存key
* @return
* 0:库存不足
* -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(" if (stock == -1) then");
sb.append(" return 1;");
sb.append(" end;");
sb.append(" if (stock > 0) then");
sb.append(" redis.call('incrby', KEYS[1], -1);");
sb.append(" return stock;");
sb.append(" end;");
sb.append(" return 0;");
sb.append("end;");
sb.append("return -1;");
STOCK_LUA = sb.toString();
}
/**
* @param key 库存key
* @param expire 库存有效时间,单位秒
* @param stockCallback 初始化库存回调函数
* @return 0:库存不足; -1:库存未初始化; 大于0:扣减库存之前的剩余库存（扣减之前剩余的库存）
*/
public long stock(String key, long expire, IStockCallback stockCallback) {
long stock = stock(key);
// 初始化库存
if (stock == UNINITIALIZED_STOCK) {
RedisLock redisLock = new RedisLock(redisTemplate, key);
try {
// 获取锁
if (redisLock.tryLock()) {
// 双重验证，避免并发时重复回源到数据库
stock = stock(key);
if (stock == UNINITIALIZED_STOCK) {
// 获取初始化库存
final int initStock = stockCallback.getStock();
// 将库存设置到redis
redisTemplate.opsForValue().set(key, initStock, expire, TimeUnit.SECONDS);
// 调一次扣库存的操作
stock = stock(key);
}
}
} catch (Exception e) {
logger.error(e.getMessage(), e);
} finally {
redisLock.unlock();
}
}
return stock;
}
/**
* 获取库存
*
* @param key 库存key
* @return 0:库存不足; -1:库存未初始化; 大于0:剩余库存
*/
public int getStock(String key) {
Integer stock = (Integer) redisTemplate.opsForValue().get(key);
return stock == null ? -1 : stock;
}
/**
* 扣库存
*
* @param key 库存key
* @return 扣减之前剩余的库存【0:库存不足; -1:库存未初始化; 大于0:扣减库存之前的剩余库存】
*/
private Long stock(String key) {
// 脚本里的KEYS参数
List<String> keys = new ArrayList<>();
keys.add(key);
// 脚本里的ARGV参数
List<String> args = new ArrayList<>();
long result = redisTemplate.execute(new RedisCallback<Long>() {
@Override
public Long doInRedis(RedisConnection connection) throws DataAccessException {
Object nativeConnection = connection.getNativeConnection();
// 集群模式和单机模式虽然执行脚本的方法一样，但是没有共同的接口，所以只能分开执行
// 集群模式
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;
}
}
源码： https://github.com/wyh-spring-ecosystem-student/spring-boot-student/tree/releases
spring-boot-student-stock-redis 工程
---------------------
作者：qq_41534566
来源：CSDN
原文：https://blog.csdn.net/qq_41534566/article/details/79219023
版权声明：本文为作者原创文章，转载请附上博文链接！
内容解析By：CSDN,CNBLOG博客文章一键转载插件
边栏推荐
- leetcode：560. 和为 K 的子数组
- Wechat applet development Basics
- leetcode：304. 二维区域和检索 - 矩阵不可变
- 2022-2028 global CAE engineering service industry research and trend analysis report
- 【TcaplusDB】祝大家端午安康！
- Want to be an equipment manager? If you meet these three conditions, you can
- Ugui slider minimum control
- vim配置与使用
- Linear and nonlinear structures
- 二叉树的层序遍历 II[层序遍历方式之一 -＞递归遍历 + level]
猜你喜欢
Basic MySQL database operations
In depth analysis of Apache bookkeeper series: Part 3 - reading principle
Concise words tell about technical people who must master basic IT knowledge and skills. Part 1
leetcode：560. 和为 K 的子数组
【线程通信】
[thread communication]
Solve the problem that the cursor flashes after clicking a point when measuring the distance in Allegro
Counter analysis of the parameter anti content in the backstage of the ＃ yyds dry goods inventory ＃ knife fight shop
問題——adb shellerror: insufficient permissions for device: verify udev rules.
Etcd tutorial - Chapter 6 etcd core API V3
随机推荐
MySQL advanced SQL statement (Part 2)
Jerry's watch stops moving [chapter]
2022-2028 global pneumatic test probe industry survey and trend analysis report
Allegro's method of setting network flying line and network color
[flutter topic] 66 diagram basic constraints box (I) yyds dry goods inventory
不同的二叉搜索樹[自下而上回溯生成樹+記憶搜索--空間換時間]
FPGA (VII) RTL code III (complex circuit design 2)
Vscode plug-in used now
Yyds dry inventory difference between bazel and gradle tools
Basic concepts of graph theory
Is the account opening of GF Securities really safe and reliable
无法定位程序输入点 [email protected]
Basic MySQL database operations
Access 500 error after modstart migrates the environment
Installation and deployment of sw-x framework
不同的二叉搜索树[自下而上回溯生成树+记忆搜索--空间换时间]
Mobaihe box, ZTE box, Migu box, Huawei box, Huawei Yuehe box, Fiberhome box, Skyworth box, Tianyi box and other operators' box firmware collection and sharing
Web GIS 航拍实现的智慧园区数字孪生应用
想当设备管理师？满足这三个报考条件就可以
2D人体姿态估计 - DeepPose