库存扣减限制

设想

有些文章说的都是商品库存的扣减或抵扣券的扣减、但是如果我有一个活动、活动中有1张券、我要限制这个活动的某张券每人限领一张、并设置月限多少等等一些库存的限制。

前期设想、我们这个最后发券的流程就是调用第三方接口给用户下发券码、假设第三方接口发券失败了、我们要回退活动的库存限制、如果下发成功应该扣减我们的库存限制。

现在关心的是活动中的券的库存限制、 现在设想的这种是在活动层面上在加库存的限制概念。

数据类型选择

Redis数据类型选择、包括一个活动是主要的信息以及券码的库存限制也是一个主要的信息、照这样说的话这两个信息要同时在Redis中或者两者能关联上、这样的话可以使用Hash的数据结构、Redis的key做为活动、hash-key做为某个库存限制、或者使用String类型也可以、将活动key和hash-key拼接在一起、这样的话也是可以只不过多个key不好管理。

库存的扣减方案

说完要使用的类型、来说下扣减的方案是预先将500的库存写到Redis当中进行库存的递减、递减完判断数量是否小于0、小于则回退库存+1、还是使用递增的方式去判断递增完数量是否大于500库存、大于则回退库存 -1

最后还是选递增的方式、假设我们的库存由后台配置、如果是扣减的方式相当于第一次添加是预设置库存为500、但是后面扣减到了450、第二天甲方说要加100库存、你直接修改第一次配置(500)的库存为600、你要拿到扣减后(450)的值以及之前配置(500)的值和拿到现在配置(600)的值、最后得出的结果是：扣减后库存+(现在配置-之前配置)然后重新加到缓存中。

但是递增的方式就不一样了、我们只需要在后台配置一个总的库存、由于库存是递增的我们后台就只需要在原有的配置库存上增加或者减少对应库存数量就可以。

简单例子

我们例子采用递增的方式来实现、通过Redis指定步进数量hincrby key field 递增的数量。

使用Java操作redisredisTemplate的increment(key,hashKey,递增的数量)方法步进。

@RequestMapping("/demo")
public String demo(@RequestParam(value = "account") long account) {
  String lockName = "sync_lock.user_" + account;
  String lockValue = UUID.randomUUID().toString();
  if (!redisLockUtils.tryLock(lockName, lockValue, 100, TimeUnit.SECONDS)) {
    return "error";
  }
  try {
    HashOperations<String, Object, Object> hashOperations = redisTemplate.opsForHash();
    String key = "activity";
    String hashKey = "total_inventory";
    Long increment = hashOperations.increment(key, hashKey, 1L);
    if (increment > 3L) {
      hashOperations.increment(key, hashKey, -1L);
      return "库存不足";
    }
  } finally {
    redisLockUtils.unlock(lockName, lockValue);
  }
  return "ok";
}

测试模拟库存扣减、当加递增的库存数量大于3L的时候将库存减一并返回库存不足、可以在上方加上分布式锁、避免了未拿到锁的线程也去做递增数据的增加、简单来说就是拦截掉了无用的后续请求。

例子进阶

上方简单的例子有一个问题、就是我有多种限制(总限制、月限、年限、日限、用户日限等.....)的时候总不能复制粘贴搞多份出来、所以后续肯定是统一封装一个工具类的、后续会将封装的上传到gitee上。

测试代码例子

@RequestMapping("/handle")
public String handle(@RequestParam("account") long account) {
  String lockName = "sync_lock.user_" + account;
  String lockValue = UUID.randomUUID().toString();
  if (!redisLockUtils.tryLock(lockName, lockValue, 100, TimeUnit.SECONDS)) {
    return "error";
  }
  try {
    String totalLimit = "totalLimit";
    String userLimit = "user_" + account;
    RedisIncreaseHelper redisIncreaseHelper = RedisIncreaseHelper.buildKey(redisTemplate, jdbcTemplate, account);
    RedisIncreaseHelper.Result result = redisIncreaseHelper
      .buildIncreaseOption("activity", totalLimit, 1L, 0L, 1000L)
      .buildIncreaseOption("activity", userLimit, 1L, 0L, 1)
      .commit();
    if (!result.getSuccess()) {
      String hashKey = result.getOption().getHashKey();
      if (totalLimit.equals(hashKey)) {
        return "总库存不足";
      } else if (userLimit.equals(hashKey)) {
        return "今天领取过喔";
      }
    }
    try {
      System.out.println("正在执行业务代码");
      TimeUnit.SECONDS.sleep(1);
      if (2 == account) {
        throw new RuntimeException("强制异常！！！");
      }
    } catch (Exception e) {
      log.error("处理[{}]、异常{}", account, e.getMessage());
      String rollbackId = result.getRollbackId();
      log.warn("rollbackId:{}", rollbackId);
      if (null != rollbackId) {
        redisIncreaseHelper.rollback(rollbackId);
      }
      return "error执行接口错误";
    }
    return "ok";
  } finally {
    redisLockUtils.unlock(lockName, lockValue);
  }
}

从以上代码看主要还是通过、RedisIncreaseHelper封装的工具类来做步进的封装、从代码可以分为在commit()步进库存和执行业务代码出现异常回滚、这个两点是核心理解的点。

第一点

先看commit()方法

public Result commit() {
  Result result = Result.success(options);
  Option option = null;
  try {
    for (Map.Entry<String, Option> entry : options.entrySet()) {
      option = entry.getValue();
      option.operate = OperateEnum.HASH_CHANGE;
      long increment = hashOperations.increment(key, option.hashKey, option.atomic);
      option.after = increment;
      option.before = increment - 1;
      option.executed = true;
      String msg = String.format("用户:[%s]、变更前后[%s]、步进值[%s]", this.accountId, option.after, option.atomic);
      if (option.min != null && option.min > increment) {
        log.error("{}、变更后的值小于于最小值、进行回滚处理。", msg);
        throw new CustomException(-1, "变更后的值小于最小值");
      }
      if (option.max != null && option.max < increment) {
        log.error("{}、变更后的值大于最大值、进行回滚处理。", msg);
        throw new CustomException(-1, "变更后的数量大于最大值");
      }
    }
  } catch (Exception e) {
    result = Result.failure(option);
    rollback();
  } finally {
    saveLog(result);
  }
  return result;
}

从上方代码来看就是遍历我们要步进的数据、无步进异常就保存变更记录、有步进异常就将result重新赋值给Controller层面提示对应的异常信息、然后在执行rollback()回滚Redis中已经步进完的数据进行 -1 操作、就不会记录报错的步进数据了。

rollback()

public void rollback() {
  for (Map.Entry<String, Option> entry : options.entrySet()) {
    if (entry.getValue().executed) {
      Option option = entry.getValue();
      hashOperations.increment(option.key, option.hashKey, -option.atomic);
    }
  }
}

saveLog()

private void saveLog(Result result) {
  if (null == result || !result.success) {
    return;
  }
  try {
    StringBuilder sql = new StringBuilder("INSERT INTO `t_activity_change_log` (`key`,`hash_key`,`operate`,`before`,`atomic`,`after`,`account_id`,`request_id`,`rollback_id`,`create_time`) VALUES(?,?,?,?,?,?,?,?,?,?)");
    for (Map.Entry<String, Option> entry : result.options.entrySet()) {
      Option option = entry.getValue();
      List<Object> params = new LinkedList<>();
      params.add(option.key);
      params.add(option.hashKey);
      params.add(option.operate.value);
      params.add(option.before);
      params.add(option.atomic);
      params.add(option.after);
      params.add(option.accountId);
      params.add(option.requestId);
      params.add(result.rollbackId);
      params.add(new Date());
      jdbcTemplate.update(sql.toString(), params.toArray());
    }
  } catch (DataAccessException e) {
    log.error("记录(成功||回滚)的日志信息异常", e);
  }

第二点

业务代码执行

try {
  System.out.println("正在执行业务代码");
  TimeUnit.SECONDS.sleep(1);
  if (2 == account) {
    throw new RuntimeException("强制异常！！！");
  }
} catch (Exception e) {
  String rollbackId = result.getRollbackId();
  log.error("处理:[{}]、异常:{}、rollbackId:{}", account, e.getMessage(),rollbackId);
  if (null != rollbackId) {
    redisIncreaseHelper.rollback(rollbackId);
  }
  return "error执行接口错误";
}

测试执行业务代码时强制异常、执行rollback(rollbackId)方法将之前commit()已经步进在Redis中的值给回滚 -1、下列是回退的代码。

public void rollback(String rollbackId) {
String sql ="select * from `t_activity_change_log` where operate = 1 and rollback_id = ? ";
  List<ActivityChangeLog> activityChangeLogs = jdbcTemplate.query(sql, new BeanPropertyRowMapper<>(ActivityChangeLog.class), rollbackId);
  if (activityChangeLogs.size() == 0) {
    log.warn("未查询到回退信息");
  }
  // 已回退数据
  Map<String, Option> handleOptions = new HashMap<>();
  // 当前回退数据
  Option option = null;
  boolean isExecuted = true;
  try {
    for (ActivityChangeLog log : activityChangeLogs) {
      option = new Option(log.getKey(), log.getHashKey(), log.getAtomic(),log.getRequestId());
     option.executed = true;
     option.before = log.getAfter();
     option.operate = OperateEnum.HASH_ROLLBACK;
     option.after = hashOperations.increment(log.getKey(), option.hashKey, -option.atomic);
     handleOptions.put(buildKeyString(log.getKey(), log.getHashKey()), option);
   }
  } catch (Exception e) {
    isExecuted = false;
    // 未回退数据
    List<ActivityChangeLog> notHandleOptions = activityChangeLogs.stream()
   .filter(log -> null == handleOptions.get(buildKeyString(log.getKey(),log.getHashKey())))
   .collect(Collectors.toList());
    log.error(String.format("回滚[%s]异常、正在回滚[%s]时异常、异常信息[%s]、需要回退[%s]条数据、已回退[%s]条数据、未回退数据[%s]、已回退数据[%s]",
                                    rollbackId, option, e, activityChangeLogs.size(), handleOptions.size(), notHandleOptions, handleOptions.values()));
    // 邮件警告等一些处理。
  }
  if (isExecuted) {
    Result result = Result.success(handleOptions);
    result.rollbackId = rollbackId;
    saveLog(result);
  }
}

业务代码异常这层面就是去查之前步进成功后的变更日志数据、将其数据回退到没步进前的值、这样就可以避免已经增加了库存、但是业务代码在执行时候执失败导致库存还占一个。

Lua脚本实现

luaCommit()

替换commit()方法中遍历步进每一个要步进的值、使用Lua脚本统一遍历步进以及回退步进值操作。

luaCommit.lua

-- 拿到json数据
local optionsStr = ARGV[1];
local accountId = ARGV[2];
-- 转成lua能处理的对象
local options = cjson.decode(optionsStr);

-- 出现异常对象
local errorOption = {};
-- 是否需要回滚数据
local errorFlag = false;
-- 遍历步进数据
for i = 1, #options do
    -- 记录当前操作
    errorOption = options[i];
    -- 记录当条数据为执行
    options[i].executed = true;
    -- 记录步进后的值
    options[i].after = redis.call('hincrby', options[i].key, options[i].hashKey, options[i].atomic);
    options[i].before = options[i].after - 1;
    options[i].operate = 'HASH_CHANGE';
    -- 判断步进后的值是否小于最小值
    if options[i].min ~= nil and options[i].min > options[i].after then
        errorFlag = true;
        redis.log(redis.LOG_NOTICE,
                "用户[" .. accountId,
                "]变更[" .. options[i].key,
                "]、[" .. options[i].hashKey,
                "]、请求流水[" .. options[i].requestId,
                "]、变更后[" .. options[i].after,
                "]、小于[" .. options[i].min,
                "]")
        break ;
    end
    -- 判断步进后的值是否大于最大值
    if options[i].max ~= nil and options[i].max < options[i].after then
        errorFlag = true;
        redis.log(redis.LOG_NOTICE,
                "用户[" .. accountId,
                "]变更[" .. options[i].key,
                "]、[" .. options[i].hashKey,
                "]、请求流水[" .. options[i].requestId,
                "]、变更后[" .. options[i].after,
                "]、大于[" .. options[i].max,
                "]")
        break ;
    end
end
-- 是否需要回滚数据
if errorFlag then
    -- 遍历回滚数据
    for i = 1, #options do
        if options[i].executed then
            redis.call('hincrby', options[i].key, options[i].hashKey, -options[i].atomic);
        end
    end
else
    -- 如不需要回滚则将errorOption置为null
    errorOption = nil
end
-- 封装数据返回
local deal = {
    errorOption = errorOption,
    options = options
}
return cjson.encode(deal)

public Result luaCommit() {
  Result result = Result.success();
  Option option = null;
  try {
    DefaultRedisScript<String> defaultRedisScript = new DefaultRedisScript<>();
    defaultRedisScript.setLocation(new ClassPathResource("luaCommit.lua"));
    defaultRedisScript.setResultType(String.class);
    // 获取lua脚本的执行结果
    JSONObject responseData = JSONUtil.parseObj(redisTemplate.opsForValue().getOperations().execute(
      defaultRedisScript,
      Collections.emptyList(),
      JSONUtil.toJsonStr(options.values()),
      String.valueOf(this.accountId)
    ));
    // 是否有异常信息
    option = JSONUtil.toBean(responseData.getJSONObject("errorOption"), Option.class);
    if (null != option) {
      throw new CustomException(-1,
                                String.format("用户:[%s]、步进key:[%s]、hashKey:[%s]、流水id:[%s]异常！！！", option.accountId, option.key, option.hashKey, option.requestId))
    }
    // 全部执行结果
    Map<String, Option> options = JSONUtil.toList(responseData.getJSONArray("options"), Option.class)
      .stream().collect(Collectors.toMap(option1 -> buildKeyString(option1.getKey(), option1.getHashKey()), Function.identity()));
    result.setOptions(options);
  } catch (Exception e) {
    log.error("步进失败:{}",e.getMessage());
    result = Result.failure(option);
  } finally {
    saveLog(result);
  }
  return result;
}

luaRollback(String rollbackId)

luaRollback.lua

-- 拿到json数据
local activityChangeLogStr = ARGV[1];
-- 要回滚的请求流水
local rollbackId = ARGV[2];
-- 转成lua能处理的对象
local activityChangeLog = cjson.decode(activityChangeLogStr);
redis.log(redis.LOG_NOTICE,"开始回滚rollbackId[" ..rollbackId,"]")

for i = 1, #activityChangeLog do
    -- 回退之前步进值
    activityChangeLog[i].after = redis.call('hincrby',activityChangeLog[i].key,activityChangeLog[i].hashKey,-activityChangeLog[i].atomic);
    activityChangeLog[i].before = activityChangeLog[i].after + 1;
    activityChangeLog[i].operate = 'HASH_ROLLBACK';
    redis.log(redis.LOG_NOTICE,
            "回滚成功[" ..activityChangeLog[i].key,
            "]、[" ..activityChangeLog[i].hashKey,
            "]、请求流水[" ..activityChangeLog[i].requestId,
            "]、回滚前[" ..activityChangeLog[i].before,
            "]、回滚后[" ..activityChangeLog[i].after,
            "]")
end
return cjson.encode(activityChangeLog)

public void luaRollback(String rollbackId) {
String sql = "select * from `t_activity_change_log` where operate = 1 and rollback_id = ?";
  List<ActivityChangeLog> activityChangeLogs = jdbcTemplate.query(sql, new BeanPropertyRowMapper<>(ActivityChangeLog.class), rollbackId);
  if (activityChangeLogs.size() == 0) {
    log.warn("未查询到回退信息");
  }
  // 处理变更日志数据
  Map<String, Option> options = new HashMap<>();
  for (ActivityChangeLog log : activityChangeLogs) {
    Option option = new Option(log.getKey(), log.getHashKey(), log.getAtomic(), log.getAccountId());
    options.put(buildKeyString(log.getKey(), log.getHashKey()), option);
  }
  boolean isExecuted = true;
  try {
    DefaultRedisScript<String> defaultRedisScript = new DefaultRedisScript<>();
    defaultRedisScript.setLocation(new ClassPathResource("luaRollback.lua"));
    defaultRedisScript.setResultType(String.class);
    // 获取lua脚本的执行结果
    JSONArray responseData = JSONUtil.parseArray(redisTemplate.opsForValue().getOperations().execute(
      defaultRedisScript,
      Collections.emptyList(),
      JSONUtil.toJsonStr(options.values()), rollbackId));
    // 全部执行结果
    options = JSONUtil.toList(responseData, Option.class)
      .stream().collect(Collectors.toMap(option -> buildKeyString(option.getKey(), option.getHashKey()), Function.identity()));
  } catch (Exception e) {
    isExecuted = false;
    log.error(String.format("回滚[%s]失败、异常信息[%s]", rollbackId, e));
    // 邮件警告等一些处理。
  }
  if (isExecuted) {
    saveLog(Result.success(options, rollbackId));
  }
}

测试代码例子

@PostConstruct
public void init() {
  redisTemplate.setKeySerializer(new StringRedisSerializer());
  redisTemplate.setValueSerializer(new StringRedisSerializer());
  redisTemplate.afterPropertiesSet();
}

@RequestMapping("/handle3")
public String handle3(@RequestParam("account") long account) {
  String lockName = "sync_lock.user_" + account;
  String lockValue = UUID.randomUUID().toString();
  if (!redisLockUtils.tryLock(lockName, lockValue, 100, TimeUnit.SECONDS)) {
    return "error";
  }
  try {
    String key = "activity3";
    String totalLimit = "totalLimit";
    String userLimit = "user_" + account;
    RedisIncreaseHelper redisIncreaseHelper = RedisIncreaseHelper.buildKey(redisTemplate, jdbcTemplate, account);
    RedisIncreaseHelper.Result result = redisIncreaseHelper
      .buildIncreaseOption(key, totalLimit, 1L, 0L, 50L)
      .buildIncreaseOption(key, userLimit, 1L, 0L, 5L)
      .luaCommit();
    if (!result.getSuccess()) {
      String hashKey = result.getOption().getHashKey();
      if (totalLimit.equals(hashKey)) {
        return "总库存不足";
      } else if (userLimit.equals(hashKey)) {
        return "今天领取过喔";
      }
    }
    try {
      System.out.println("正在执行业务代码");
      TimeUnit.SECONDS.sleep(1);
      if (2 == account) {
        throw new RuntimeException("强制异常！！！");
      }
    } catch (Exception e) {
      String rollbackId = result.getRollbackId();
      log.error("处理[{}]、rollbackId:{}、异常{}", account, rollbackId, e.getMessage());
      if (null != rollbackId) {
        redisIncreaseHelper.luaRollback(rollbackId);
      }
      return "error执行接口错误";
    }
    return "ok";
  } finally {
    redisLockUtils.unlock(lockName, lockValue);
  }
}

小结

注意：使用lua的时候redisTemplate的要重新设置一下redisTemplate.setValueSerializer(new StringRedisSerializer());value值的序列化器、json的序列化器会导致传参报错、或者直接注入StringRedisTemplate使用。

注意： redis.log(redis.LOG_NOTICE,.....) 在lua脚本中这个作用是在redis记录日志信息、需要在redis.conf文件中logfile:xxxxx配置日志保存位置文件地址。

大致流程图

最终版本测试

总库存[totalLimit] = 10、月限[month_202211] = 3、单用户[user_*] = 2、随机用户测试、启动两个服务、进行nginx负载、使用jmeter工具测试。

猜想：如三个用户各一张、再一次拿就会超过月限、或某个用户拿多张、其他用户拿一张、再一次拿就会超过月限、代码中强制用户2处理业务逻辑强制异常了、所以用户2永远都是0。

测试入口代码

@PostConstruct
public void init() {
  redisTemplate.setKeySerializer(new StringRedisSerializer());
  redisTemplate.setValueSerializer(new StringRedisSerializer());
  redisTemplate.afterPropertiesSet();
}
@RequestMapping("/handle3")
public String handle3(@RequestParam("account") long account) {
  String lockName = "sync_lock.user_" + account;
  String lockValue = UUID.randomUUID().toString();
  if (!redisLockUtils.tryLock(lockName, lockValue, 100, TimeUnit.SECONDS)) {
    return "error";
  }
  try {
    String key = "activity3";
    String totalLimit = "totalLimit";
    String monthLimit = "month_202211";
    String userLimit = "user_" + account;
    RedisIncreaseHelper redisIncreaseHelper = RedisIncreaseHelper.buildKey(redisTemplate, jdbcTemplate, account);
    RedisIncreaseHelper.Result result = redisIncreaseHelper
      .buildIncreaseOption(key, totalLimit, 1L, 0L, 10L)
      .buildIncreaseOption(key, userLimit, 1L, 0L, 2L)
      .buildIncreaseOption(key, monthLimit, 1L, 0L, 3L)
      .luaCommit();
    if (!result.getSuccess()) {
      String hashKey = result.getOption().getHashKey();
      if (totalLimit.equals(hashKey)) {
        return "总库存不足";
      } else if (monthLimit.equals(hashKey)) {
        return "月库存不足";
      }else if (userLimit.equals(hashKey)) {
        return "今天领取过喔";
      }
    }
    try {
      System.out.println("正在执行业务代码");
      TimeUnit.SECONDS.sleep(1);
      if (2 == account) {
        throw new RuntimeException("强制异常！！！");
      }
    } catch (Exception e) {
      String rollbackId = result.getRollbackId();
      log.error("处理[{}]、rollbackId:{}、异常{}", account, rollbackId, e.getMessage());
      if (null != rollbackId) {
        redisIncreaseHelper.luaRollback(rollbackId);
      }
      return "error执行接口错误";
    }
    return "ok";
  } finally {
    redisLockUtils.unlock(lockName, lockValue);
  }
}

测试结果

由于上方代码没有执行到用户2的强制异常、所以增加一个库存手动模拟用户2调用抢券、请求地址http://localhost:8089/handle3?account=2

由于用户2步进时是还有1个库存的、所以步进的日志信息是存在数据库当中的、但是执行业务代码时出现强制性异常后就想之前步进的值回滚 -1 了、回滚成功后的日志也记录在数据库中。