电商网站如何进行库存同步处理Redis+Lua

电商网站库存模块

库存表包含了商品的sku，商品类型，商品款号，颜色，尺码，库存数，版本号，创建时间，修改时间。

商品类型，可根据商品分为，普通商品，赠品，内卖商品，预售商品等

库存表结构

@Data
public class ProductStock extends OrderEntity<String> {private static final long serialVersionUID = 6324321924144806460L;/*** sku*/private String sku;/*** good type*/private GoodsType goodsType;/*** 款号*/private String sn;/*** 颜色*/private String color;/*** 尺码*/private String size;/*** 库存*/private Integer stock;@Versionprivate Long version;private Date createdDate;private Date lastModifiedDate;
}

有库存表就有库存操作表，记录每次出库入库的日志。

库存操作记录表

@Data
public class ProductStockLog {private static final long serialVersionUID = -5035787394251728152L;/*** 类型*/public enum Type {/** 入库 */stockIn,/** 出库 */stockOut}/** 类型 */private ProductStockLog.Type type;private String sku;private GoodsType goodsType;/** 入库数量 */private Integer inQuantity;/** 出库数量 */private Integer outQuantity;/** 当前库存 */private Integer stock;/** 操作员 */private String operator;/** 备注 */private String memo;/** 商品 */private String productName;@Versionprivate Long version;private Date createdDate;private Date lastModifiedDate;
}

基础的实体创建好了，那就需要具体的操作库存了，库存的扣减一般分为下单扣库存，支付扣库存等，可根据项目需要在合适的场景下进行库存操作。

首先下单的时候，肯定需要先检查库存。

检查库存

通过sku和商品的类型查询库存，返回Integer类型

    public BaseResponse<Integer> getStockNumBySku(String sku,GoodsType type) {try {String key = stockRequest.getType().name() + "_" + stockRequest.getSku();log.info("init key:{}", key);Long stock = redisStockService.getStock(key, 60 * 60, (IStockCallback) productStockService);if (stock == null || stock.equals(UNINITIALIZED_STOCK)) {log.info("没有该商品库存,请先维护商品库存数据!");return BaseResponse.exceptionResponse("500", "没有该商品库存,请先维护商品库存数据!");} else {return BaseResponse.successResponse(Integer.parseInt(String.valueOf(stock)));}} catch (Exception e) {log.error("服务器端查询库存失败", e);return BaseResponse.exceptionResponse("500", "服务器端查询库存失败");}}

通过redis + lua 保证库存原子性操作。
检查库存完毕后，就是调整库存了，一般情况下并发都是发生在扣库存的场景下，我们先创建一个库存调整请求类

@Data
@ApiModel(description = "调整库存请求类")
public class AdjustStockRequest {@ApiModelProperty(value = "sku", name = "sku", example = "sku")private String sku;@ApiModelProperty(value = "goodsType")private GoodsType goodsType;@ApiModelProperty(value = "stock", name = "增、减的库存量（负数表示减库存）", example = "0")private Integer stock;@ApiModelProperty(value = "商品SKU名称", name = "productName", example = "LV")private String productName;}

以及库存操作状态的枚举类型

public enum StockOperationStatEnum {SUCCESS(1, "库存操作成功"),STOCK_UNLIMITED(-1,"库存不限"),STOCK_NOT_ENOUGH(-2, "库存不足"),STOCK_UNINITIALIZED(-3, "库存未初始化"),FAILED(-4, "库存操作失败");private long state;private String stateInfo;StockOperationStatEnum(long state, String stateInfo) {this.state = state;this.stateInfo = stateInfo;}public long getState() {return state;}public String getStateInfo() {return stateInfo;}public static StockOperationStatEnum stateOf(long index){for (StockOperationStatEnum state : values()){if (state.getState() == index){return state;}}return null;}
}

调整库存

    public BaseResponse<Boolean> adjustStock(@RequestBody AdjustStockRequest stockRequest) {// TODO 省略了参数的非空校验String key = stockRequest.getGoodsType() + "_" + stockRequest.getSku();StockOperationStatEnum resultEnum = productStockService.adjustStock(key, stockRequest.getStock());if (resultEnum == StockOperationStatEnum.SUCCESS) {ProductStockLog productStockLog = new ProductStockLog();productStockLog.setSku(stockRequest.getSku());productStockLog.setGoodsType(stockRequest.getGoodsType());productStockLog.setType(stockRequest.getStock() > 0 ? ProductStockLog.Type.stockIn : ProductStockLog.Type.stockOut);productStockLog.setInQuantity(stockRequest.getStock() > 0 ? stockRequest.getStock() : 0);productStockLog.setOutQuantity(stockRequest.getStock() < 0 ? Math.abs(stockRequest.getStock()) : 0);Long stock = redisStockService.getStock(key, 60 * 60, (IStockCallback) productStockService);if (stock == null || stock.equals(UNINITIALIZED_STOCK)) {stock = Long.valueOf(0);}//当前库存productStockLog.setStock(Integer.parseInt(String.valueOf(stock)));productStockLog.setOperator("customer");productStockLog.setProductName(stockRequest.getProductName());productStockLogService.saveProductStockLog(productStockLog);//  调整库存后更改商品假库存ProductStockVo productStockVo = new ProductStockVo();productStockVo.setStock(Integer.parseInt(String.valueOf(stock)));productStockVo.setGoodsType(stockRequest.getGoodsType());productStockVo.setSku(stockRequest.getSku());stockSendGoods.send(GsonUtil.toJson(productStockVo));return BaseResponse.successResponse(true);}return BaseResponse.exceptionResponse("500",false);}

整个库存的操作我们都是放到了redis中，通过redis+lua保证库存操作的原子性，最终在完成库存调整后，通过ProductStockVo 对象，将库存调整的信息通过MQ同步到我们库存的WMS系统。

Redis+Lua

Lua 嵌入 Redis 优势:

减少网络开销: 不使用 Lua 的代码需要向 Redis 发送多次请求, 而脚本只需一次即可, 减少网络传输;
原子操作: Redis 将整个脚本作为一个原子执行, 无需担心并发, 也就无需事务;
复用: 脚本会永久保存 Redis 中, 其他客户端可继续使用.

@Service
@Slf4j
public class RedisStockService {/*** Redis 客户端*/@Autowiredprivate RedisTemplate<String, Object> redisTemplate;public static final String REDIS_STOCK_KEY_PREFIX = "redis_key:stock:";public static final String REDIS_INIT_STOCK_KEY_PREFIX = "redis_key:stock_init:";/*** 库存未初始化*/public static final Long UNINITIALIZED_STOCK = -3L;/*** 执行扣库存的脚本*/private 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 (num < 0) then");sb.append("       if (stock >= math.abs(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 redis.call('incrby', KEYS[1], 0 + num);");sb.append("end;");sb.append("return -3;");STOCK_LUA = sb.toString();log.info("init stock_lua script:{}", STOCK_LUA);}/*** @param key           库存key* @param expire        库存有效时间,单位秒* @param num           扣减数量* @param stockCallback 初始化库存回调函数* @return -2:库存不足; -1:不限库存; 大于等于0:扣减库存之后的剩余库存*/public Long adjustStock(String key, long expire, int num, IStockCallback stockCallback) {if (StringUtils.isEmpty(key)) {key = "";}if (StringUtils.countMatches(key, REDIS_STOCK_KEY_PREFIX) <= 0) {key = REDIS_STOCK_KEY_PREFIX + key;}//redis lua脚本修改库存long stock = this.adjustStock(key, num);// 如果没有初始库存if (stock == UNINITIALIZED_STOCK) {RedisLock redisLock = new RedisLock(redisTemplate, key);try {// 获取锁if (redisLock.tryLock()) {// 双重验证，避免并发时重复回源到数据库stock = this.adjustStock(key, num);if (stock == UNINITIALIZED_STOCK) {// 获取初DB库存final Long initStock = stockCallback.getDataBaseStock(key);if (initStock != null && initStock >= 0) {// 将库存设置到redisredisTemplate.opsForValue().set(key, initStock, expire, TimeUnit.SECONDS);// 调整库存的操作stock = this.adjustStock(key, num);}}}} catch (Exception e) {log.error(e.getMessage(), e);} finally {redisLock.unlock();}}return stock;}/*** 获取库存** @param key 库存key* @return -1:不限库存; 大于等于0:剩余库存*/public Long getStock(String key, long expire, IStockCallback stockCallback) {if (StringUtils.isEmpty(key)) {key = "";}if (StringUtils.countMatches(key, REDIS_STOCK_KEY_PREFIX) <= 0) {key = REDIS_STOCK_KEY_PREFIX + key;}Integer stockCache = (Integer) redisTemplate.opsForValue().get(key);Long stock = null;if (stockCache == null || stockCache.longValue() == UNINITIALIZED_STOCK) {RedisLock redisLock = new RedisLock(redisTemplate, key);try {// 获取锁if (redisLock.tryLock()) {if (stockCache == null || stockCache.longValue() == UNINITIALIZED_STOCK) {// 获取DB库存final Long initStock = stockCallback.getDataBaseStock(key);if (initStock != null && initStock >= 0) {// 将库存设置到redisredisTemplate.opsForValue().set(key, initStock, expire, TimeUnit.SECONDS);}stock = initStock;}}} catch (Exception e) {log.error(e.getMessage(), e);} finally {redisLock.unlock();}} else {stock = stockCache.longValue();}return stock;}public void setRedisKeyValue(String key, Object stock) {if (StringUtils.isEmpty(key)) {key = "";}if (StringUtils.countMatches(key, REDIS_STOCK_KEY_PREFIX) <= 0) {key = REDIS_STOCK_KEY_PREFIX + key;}redisTemplate.opsForValue().set(key, stock, 60 * 60, TimeUnit.SECONDS);}/*** 扣库存** @param key 库存key* @param num 扣减库存数量* @return 扣减之后剩余的库存【-3:库存未初始化; -2:库存不足; -1:不限库存; 大于等于0:扣减库存之后的剩余库存】*/private Long adjustStock(String key, int num) {if (StringUtils.isEmpty(key)) {key = "";}if (StringUtils.countMatches(key, REDIS_STOCK_KEY_PREFIX) <= 0) {key = REDIS_STOCK_KEY_PREFIX + key;}// 脚本里的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>() {@Overridepublic 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;}}

getStock获取库存的时候，如果是首次获取，库存并没存在缓存中，或者库存尚未初始化时，通过redis锁防止分布式场景中并发操作，DB获取库存放到redis中。
adjustStock如果调整库存失败，将会通过回调函数stockCallback，初始化当前库存，并且再将库存设置到redis中。

回调函数初始化当前库存

public interface IStockCallback {/*** 获取库存* @return*/Long getDataBaseStock(String sku);
}

分布式环境下，我们需要一个分布式锁来控制只能有一个服务去初始化库存，因此在对库存进行操作的时候，优先判断redis是否获取到了锁tryLock

redis分布式锁

/*** Redis分布式锁* 使用 SET resource-name anystring NX EX max-lock-time 实现* EX seconds — 以秒为单位设置 key 的过期时间；* PX milliseconds — 以毫秒为单位设置 key 的过期时间；* NX — 将key 的值设为value ，当且仅当key 不存在，等效于 SETNX。* XX — 将key 的值设为value ，当且仅当key 存在，等效于 SETEX。* <p>* 命令 SET resource-name anystring NX EX max-lock-time 是一种在 Redis 中实现锁的简单方法。* <p>* 客户端执行以上的命令：* <p>* 如果服务器返回 OK ，那么这个客户端获得锁。* 如果服务器返回 NIL ，那么客户端获取锁失败，可以在稍后再重试。**/
@Slf4j
public class RedisLock {private static Logger logger = LoggerFactory.getLogger(RedisLock.class);private RedisTemplate<String, Object> redisTemplate;/*** 将key 的值设为value ，当且仅当key 不存在，等效于 SETNX。*/public static final String NX = "NX";/*** seconds — 以秒为单位设置 key 的过期时间，等效于EXPIRE key seconds*/public static final String EX = "EX";/*** 调用set后的返回值*/public static final String OK = "OK";/*** 默认请求锁的超时时间(ms 毫秒)*/private static final long TIME_OUT = 100;/*** 默认锁的有效时间(s)*/public static final int EXPIRE = 60;/*** 解锁的lua脚本*/public static final String UNLOCK_LUA;static {StringBuilder sb = new StringBuilder();sb.append("if redis.call(\"get\",KEYS[1]) == ARGV[1] ");sb.append("then ");sb.append("    return redis.call(\"del\",KEYS[1]) ");sb.append("else ");sb.append("    return 0 ");sb.append("end ");UNLOCK_LUA = sb.toString();log.debug("init UNLOCK_LUA:{}", UNLOCK_LUA);}/*** 锁标志对应的key*/private String lockKey;/*** 记录到日志的锁标志对应的key*/private String lockKeyLog = "";/*** 锁对应的值*/private String lockValue;/*** 锁的有效时间(s)*/private int expireTime = EXPIRE;/*** 请求锁的超时时间(ms)*/private long timeOut = TIME_OUT;/*** 锁标记*/private volatile boolean locked = false;final Random random = new Random();/*** 使用默认的锁过期时间和请求锁的超时时间** @param redisTemplate* @param lockKey       锁的key（Redis的Key）*/public RedisLock(RedisTemplate<String, Object> redisTemplate, String lockKey) {this.redisTemplate = redisTemplate;this.lockKey = lockKey + "_lock";}/*** 使用默认的请求锁的超时时间，指定锁的过期时间** @param redisTemplate* @param lockKey       锁的key（Redis的Key）* @param expireTime    锁的过期时间(单位：秒)*/public RedisLock(RedisTemplate<String, Object> redisTemplate, String lockKey, int expireTime) {this(redisTemplate, lockKey);this.expireTime = expireTime;}/*** 使用默认的锁的过期时间，指定请求锁的超时时间** @param redisTemplate* @param lockKey       锁的key（Redis的Key）* @param timeOut       请求锁的超时时间(单位：毫秒)*/public RedisLock(RedisTemplate<String, Object> redisTemplate, String lockKey, long timeOut) {this(redisTemplate, lockKey);this.timeOut = timeOut;}/*** 锁的过期时间和请求锁的超时时间都是用指定的值** @param redisTemplate* @param lockKey       锁的key（Redis的Key）* @param expireTime    锁的过期时间(单位：秒)* @param timeOut       请求锁的超时时间(单位：毫秒)*/public RedisLock(RedisTemplate<String, Object> redisTemplate, String lockKey, int expireTime, long timeOut) {this(redisTemplate, lockKey, expireTime);this.timeOut = timeOut;}/*** 自旋锁* 尝试获取锁 超时返回** @return*/public boolean tryLock() {// 生成随机keylockValue = UUID.randomUUID().toString();// 请求锁超时时间，纳秒long timeout = timeOut * 1000000;// 系统当前时间，纳秒long nowTime = System.nanoTime();while ((System.nanoTime() - nowTime) < timeout) {if (OK.equalsIgnoreCase(this.set(lockKey, lockValue, expireTime))) {locked = true;// 上锁成功结束请求log.debug("获取redis锁:key{},value:{},expireTime:{}", lockKey, lockValue, expireTime);return locked;}// 每次请求等待一段时间seleep(10, 50000);}return locked;}/*** 尝试获取锁 立即返回** @return 是否成功获得锁*/public boolean lock() {lockValue = UUID.randomUUID().toString();//不存在则添加 且设置过期时间（单位ms）String result = set(lockKey, lockValue, expireTime);locked = OK.equalsIgnoreCase(result);return locked;}/*** 以阻塞方式的获取锁** @return 是否成功获得锁*/public boolean lockBlock() {lockValue = UUID.randomUUID().toString();while (true) {//不存在则添加 且设置过期时间（单位ms）String result = set(lockKey, lockValue, expireTime);if (OK.equalsIgnoreCase(result)) {locked = true;return locked;}// 每次请求等待一段时间seleep(10, 50000);}}/*** 解锁* <p>* 可以通过以下修改，让这个锁实现更健壮：* <p>* 不使用固定的字符串作为键的值，而是设置一个不可猜测（non-guessable）的长随机字符串，作为口令串（token）。* 不使用 DEL 命令来释放锁，而是发送一个 Lua 脚本，这个脚本只在客户端传入的值和键的口令串相匹配时，才对键进行删除。* 这两个改动可以防止持有过期锁的客户端误删现有锁的情况出现。*/public Boolean unlock() {// 只有加锁成功并且锁还有效才去释放锁// 只有加锁成功并且锁还有效才去释放锁if (locked) {return (Boolean) redisTemplate.execute(new RedisCallback<Boolean>() {@Overridepublic Boolean doInRedis(RedisConnection connection) throws DataAccessException {Object nativeConnection = connection.getNativeConnection();Long result = 0L;List<String> keys = new ArrayList<>();keys.add(lockKey);List<String> values = new ArrayList<>();values.add(lockValue);// 集群模式if (nativeConnection instanceof JedisCluster) {result = (Long) ((JedisCluster) nativeConnection).eval(UNLOCK_LUA, keys, values);}// 单机模式if (nativeConnection instanceof Jedis) {result = (Long) ((Jedis) nativeConnection).eval(UNLOCK_LUA, keys, values);}if (result == 0 && !StringUtils.isEmpty(lockKeyLog)) {logger.info("Redis分布式锁，解锁{}失败！解锁时间：{}", lockKeyLog, System.currentTimeMillis());}locked = result == 0;return result == 1;}});}return true;}/*** 获取锁状态** @return* @Title: isLock*/public boolean isLock() {return locked;}/*** 重写redisTemplate的set方法* <p>* 命令 SET resource-name anystring NX EX max-lock-time 是一种在 Redis 中实现锁的简单方法。* <p>* 客户端执行以上的命令：* <p>* 如果服务器返回 OK ，那么这个客户端获得锁。* 如果服务器返回 NIL ，那么客户端获取锁失败，可以在稍后再重试。** @param key     锁的Key* @param value   锁里面的值* @param seconds 过去时间（秒）* @return*/private String set(final String key, final String value, final long seconds) {Assert.isTrue(!StringUtils.isEmpty(key), "key不能为空");return (String) redisTemplate.execute(new RedisCallback<String>() {@Overridepublic String doInRedis(RedisConnection connection) throws DataAccessException {Object nativeConnection = connection.getNativeConnection();String result = null;if (nativeConnection instanceof JedisCommands) {result = ((JedisCommands) nativeConnection).set(key, value, NX, EX, seconds);}if (!StringUtils.isEmpty(lockKeyLog) && !StringUtils.isEmpty(result)) {logger.info("获取锁{}的时间：{}", lockKeyLog, System.currentTimeMillis());}return result;}});}/*** @param millis 毫秒* @param nanos  纳秒* @Title: seleep* @Description: 线程等待时间*/private void seleep(long millis, int nanos) {try {Thread.sleep(millis, random.nextInt(nanos));} catch (InterruptedException e) {logger.info("获取分布式锁休眠被中断：", e);}}public String getLockKeyLog() {return lockKeyLog;}public void setLockKeyLog(String lockKeyLog) {this.lockKeyLog = lockKeyLog;}public int getExpireTime() {return expireTime;}public void setExpireTime(int expireTime) {this.expireTime = expireTime;}public long getTimeOut() {return timeOut;}public void setTimeOut(long timeOut) {this.timeOut = timeOut;}}

这样就大功告成了吗？其实细心的同学已经发现了在我们代码中有这么一段Integer stockCache = (Integer) redisTemplate.opsForValue().get(key);，那是因为我在向redis放库存的时候存放的是Long类型，所以在这里需要强转一下，使用RedisTemplate的时候默认的序列化反序列方式为JdkSerializationRedisSerializer，如果我们只是存一下参数倒还无所谓，如果存的是序列化的对象，那么反序列化拿到的key就不再是我们需要的key了。当然在这个项目中，如果我们约定了redis，key的类型是Integer，那么get的时候就不需要转换。但是我们在set的时候是用的Long类型，那么通过Integer强转的时候，肯定会报转换错误。所以需要我们自定义一下redis序列化方式，指定redis 的key和value使用什么类型。

自定义序列化方式

@Configuration
public class RedisConfig {/*** 重写Redis序列化方式，使用Json方式:* 当我们的数据存储到Redis的时候，我们的键（key）和值（value）都是通过Spring提供的Serializer序列化到数据库的。RedisTemplate默认使用的是JdkSerializationRedisSerializer，StringRedisTemplate默认使用的是StringRedisSerializer。* Spring Data JPA为我们提供了下面的Serializer：* GenericToStringSerializer、Jackson2JsonRedisSerializer、JacksonJsonRedisSerializer、JdkSerializationRedisSerializer、OxmSerializer、StringRedisSerializer。* 在此我们将自己配置RedisTemplate并定义Serializer。** @param redisConnectionFactory* @return*/@Beanpublic RedisTemplate<String, Object> redisTemplate(RedisConnectionFactory redisConnectionFactory) {RedisTemplate<String, Object> redisTemplate = new RedisTemplate<>();redisTemplate.setConnectionFactory(redisConnectionFactory);Jackson2JsonRedisSerializer<Object> jackson2JsonRedisSerializer = new Jackson2JsonRedisSerializer<Object>(Object.class);ObjectMapper om = new ObjectMapper();om.setVisibility(PropertyAccessor.ALL, JsonAutoDetect.Visibility.ANY);om.enableDefaultTyping(ObjectMapper.DefaultTyping.NON_FINAL);jackson2JsonRedisSerializer.setObjectMapper(om);// 设置值（value）的序列化采用Jackson2JsonRedisSerializer。redisTemplate.setValueSerializer(jackson2JsonRedisSerializer);redisTemplate.setHashValueSerializer(jackson2JsonRedisSerializer);// 设置键（key）的序列化采用StringRedisSerializer。redisTemplate.setKeySerializer(new StringRedisSerializer());redisTemplate.setHashKeySerializer(new StringRedisSerializer());redisTemplate.afterPropertiesSet();return redisTemplate;}
}

ok。这样在redis放值取值时，如果类型不一致，进行转换的时候就不会出错了。

当然这仅仅是商城端扣减库存的操作，在上边的代码中我们并没有太多的针对库存系统（类似WMS）同步的问题，只是在调整的时候通过MQ将商城端的库存同步过去了。如果MQ失败了会怎样，岂不是从一开始就错下去了。
其实在实际项目中，这种这种商城端库存推送我们一般称之为“增量库存”，增量库存失败的情况下，系统自动调用同步全量库存。同时为了尽可能保证完全商城商品库存和仓库商品库存同步，一般需要系统定时的去获取仓库的全量库存，来尽可能保证仓库库存与商城库存一致。

之前看过一篇关于中台程序的电商更新文档，里边一些业务实现方案还是可以参考一下的：比如这个增量同步库存扣减失败系统库存同步处理