Skip to main content

物品发放幂等设计

About 5 min全民制作人iKun

物品发放幂等设计

幂等设计

在我们的系统设计中，关于用户和物品主要由三张表：

  1. user表
  2. item_config表：可表示改名卡或者佩戴的徽章
  3. user_backpack表：用于关联user和item_config表，使得user和item_config为多对多的关系

如何确保给用户发放物品，最终的结果不会多发？

mideng.svg

在分布式场景下，这些交互的事件不在一个事务里，那么就是不可靠的，一定会出现请求重试等情况，如果不去保证幂等性，那么就会出现超发的情况。

幂等边界设计：

  • 购买渠道：一个订单号只会发放一次
  • 注册渠道：一个uid只会发放一次
  • 点赞渠道：一条消息只会发放一次(如果一条消息被超过10人点赞，那么我们可以发一个徽章给予鼓励)

幂等号设计=itemId+source+bussinessId

具体代码：

@Override
public void acquireItem(Long uid, Long itemId, IdempotentEnum idempotentEnum, String businessId) {
    String idempotentId = getIdempotent(itemId, idempotentEnum, businessId);
    RLock lock = redissonClient.getLock("acquireItem" + idempotentId);
    boolean flag = lock.tryLock();
    AssertUtil.isTrue(flag, "请求太频繁了～");
    try {
        UserBackpack userBackpack = getByIdempotent(idempotentId);
        if (Objects.nonNull(userBackpack)) { // 已经获取过了
            return;
        }
        // 发放物品
        UserBackpack insertUserBackPack = new UserBackpack();
        insertUserBackPack.setUid(uid);
        insertUserBackPack.setItemId(itemId);
        insertUserBackPack.setStatus(YesOrNoEnum.NO.getStatus());
        insertUserBackPack.setIdempotent(idempotentId);
        save(insertUserBackPack);
    } finally {
        lock.unlock();
    }
}

Redis son加锁模版：

   RLock lock = redissonClient.getLock("name");
    boolean flag = lock.tryLock();
    AssertUtil.isTrue(flag, "请求太频繁了～");
    try {
        xxx
    } finally {
        lock.unlock();
    }

分布式锁封装-编程式

/**
 * @author houyunfei
 */
@Slf4j
public class LockService {
    @Resource
    private RedissonClient redissonClient;

    public <T> T executeWithLock(String key, int waitTime, TimeUnit unit, Supplier<T> supplier) throws InterruptedException {
        RLock lock = redissonClient.getLock(key);
        boolean flag = lock.tryLock(waitTime, unit);
        if (!flag) {
            throw new BusinessException(CommonErrorEnum.LOCK_LIMIT);
        }
        try {
            return supplier.get();
        } finally {
            lock.unlock();
        }
    }

    public <T> T executeWithLock(String key, Supplier<T> supplier) throws InterruptedException {
        return executeWithLock(key, -1, TimeUnit.MILLISECONDS, supplier);
    }

    public <T> T executeWithLock(String key, Runnable runnable) throws InterruptedException {
        return executeWithLock(key, -1, TimeUnit.MILLISECONDS, () -> {
            runnable.run();
            return null;
        });
    }
}

修改原来的代码，现在会简介很多：

@Override
public void acquireItem(Long uid, Long itemId, IdempotentEnum idempotentEnum, String businessId) throws InterruptedException {
    String idempotentId = getIdempotent(itemId, idempotentEnum, businessId);
    lockService.executeWithLock("acquireItem" + idempotentId, () -> {
        UserBackpack userBackpack = getByIdempotent(idempotentId);
        if (Objects.nonNull(userBackpack)) { // 已经获取过了
            return;
        }
        // 发放物品
        UserBackpack insertUserBackPack = new UserBackpack();
        insertUserBackPack.setUid(uid);
        insertUserBackPack.setItemId(itemId);
        insertUserBackPack.setStatus(YesOrNoEnum.NO.getStatus());
        insertUserBackPack.setIdempotent(idempotentId);
        save(insertUserBackPack);
    });
}

分布式锁封装-注解式

定义分布式锁注解：

@Retention(RetentionPolicy.RUNTIME)
@Target(ElementType.METHOD)
public @interface RedissonLock {
    /**
     * key的前缀，默认取方法名
     *
     * @return
     */
    String prefix() default "";

    /**
     * 支持spring EL表达式
     *
     * @return
     */
    String key();

    /**
     * 等待锁的排队事件，默认-1，快速失败
     *
     * @return
     */
    int waitTime() default -1;

    /**
     * 时间单位，默认毫秒
     *
     * @return
     */
    TimeUnit unit() default TimeUnit.MILLISECONDS;

}

定义分布式锁切面：

@Component
@Aspect
@Order(0) // 保证在事务之前执行，分布式锁需要在事务之前执行
public class RedissonLockAspect {

    @Resource
    private LockService lockService;

    @Around("@annotation(com.yunfei.chat.common.annotation.RedissonLock)")
    public Object around(ProceedingJoinPoint joinPoint) throws Throwable {
        Signature signature = joinPoint.getSignature();
        MethodSignature signature1 = (MethodSignature) signature;
        Method method = signature1.getMethod();
        RedissonLock redissonLock = method.getAnnotation(RedissonLock.class);
        String prefix = StrUtil.isBlank(redissonLock.prefix()) ? SpElUtils.getMethodKey(method) : redissonLock.prefix();
        String key = SpElUtils.parseSpEl(method, joinPoint.getArgs(), redissonLock.key());
        return lockService.executeWithLock(prefix + ":" + key, redissonLock.waitTime(), redissonLock.unit(), joinPoint::proceed);
    }
}

分布式锁的切面要设置优先级为0，确保比事务注解先执行

如果先执行事务，后执行锁，那么锁提交了，下一个人进来了，这时候上一个事务还没有提交，这样就会导致脏读，锁就没你意义了。

正确流程：

  1. 加锁
  2. 开启事务
  3. 结束事务
  4. 解锁

测试使用：

在修改名字的地方，既有事务，又有锁：

@Override
@Transactional(rollbackFor = Exception.class)
@RedissonLock(key = "#uid")
public void modifyName(Long uid, String name) {
    User user = baseMapper.getUserByName(name);
    AssertUtil.isEmpty(user, "用户名已存在，请换一个～");
    UserBackpack userBackpack = userBackpackService.getFirstValidItem(uid, ItemEnum.MODIFY_NAME_CARD.getId());
    AssertUtil.isNotEmpty(userBackpack, "您没有改名卡了～");
    // 使用改名卡
    boolean res = userBackpackService.useItem(userBackpack);
    if (res) {
        baseMapper.modifyName(uid, name);
    }
}

我们没有指定prefix，于是就是拿方法名做prefix：

image-20240519094713698

我们指定的key式通过EL表达式获取uid，于是拿到了

image-20240519094832002

接着往下走，发现也加锁成功了：

image-20240519094921628

EL表达式工具类：

用来解析输入参数里的EL表达式，其实就是将

  • 参数的名称作为key
  • 参数的值作为value
  • 例如key为uid，值为20001

然后Spring对其进行解析，将#uid替换为20001

public class SpElUtils {
    private static final ExpressionParser parser = new SpelExpressionParser();
    private static final DefaultParameterNameDiscoverer parameterNameDiscoverer = new DefaultParameterNameDiscoverer();

    public static String parseSpEl(Method method, Object[] args, String spEl) {
        String[] params = Optional.ofNullable(parameterNameDiscoverer.getParameterNames(method)).orElse(new String[]{});// 解析参数名
        EvaluationContext context = new StandardEvaluationContext();// el解析需要的上下文对象
        for (int i = 0; i < params.length; i++) {
            context.setVariable(params[i], args[i]);// 所有参数都作为原材料扔进去
        }
        Expression expression = parser.parseExpression(spEl);
        return expression.getValue(context, String.class);
    }

    public static String getMethodKey(Method method) {
        return method.getDeclaringClass() + "#" + method.getName();
    }
}

切面失效问题

我们把发放物品的逻辑修改如下，使用自己的RedissonLock注解加锁，并且加上事务

@Override
public void acquireItem(Long uid, Long itemId, IdempotentEnum idempotentEnum, String businessId) throws InterruptedException {
    String idempotentId = getIdempotent(itemId, idempotentEnum, businessId);
    //第三种方式：使用自定义的注解
    oAcquireItem(uid, itemId, idempotentId);
}

@RedissonLock(key = "#idempotentId", waitTime = 5000)
@Transactional
public void doAcquireItem(Long uid, Long itemId, String idempotentId) {
    UserBackpack userBackpack = getByIdempotent(idempotentId);
    if (Objects.nonNull(userBackpack)) { // 已经获取过了
        return;
    }
    // 发放物品
    UserBackpack insertUserBackPack = new UserBackpack();
    insertUserBackPack.setUid(uid);
    insertUserBackPack.setItemId(itemId);
    insertUserBackPack.setStatus(YesOrNoEnum.NO.getStatus());
    insertUserBackPack.setIdempotent(idempotentId);
    save(insertUserBackPack);
}

问题出现：

在这段代码中，使用了自定义注解@RedissonLock来实现分布式锁的功能，用于控制并发访问。在调用doAcquireItem方法时，会先获取锁，然后执行方法内部的逻辑。

在同类调用的情况下，可能会出现切面问题，主要是因为Spring AOP默认情况下不会拦截同类中的方法调用。所以如果doAcquireItem方法直接被调用而不是通过代理对象调用，那么@RedissonLock注解将不会生效，也就不会获取到分布式锁。

为了解决这个问题，可以获取代理对象，使用代理对象调用，确保注解生效。

简单来说：同类调用切面不生效

自己注入自己-循环依赖

自己注入自己会有循环依赖问题，可以加一个@Lazy来解决

    @Resource
    @Lazy
    private UserBackpackServiceImpl userBackpackService;

@Override
public void acquireItem(Long uid, Long itemId, IdempotentEnum idempotentEnum, String businessId) throws InterruptedException {
    String idempotentId = getIdempotent(itemId, idempotentEnum, businessId);
    //第三种方式：使用自定义的注解
    userBackpackService.doAcquireItem(uid, itemId, idempotentId);
    doAcquireItem(uid, itemId, idempotentId);

}

使用代理

((UserBackpackServiceImpl) AopContext.currentProxy()).doAcquireItem(uid, itemId, idempotentId);
页脚信息-备案信息
Copyright © 2024 全民制作人iKun MIT Licensed