redisson 分布式锁 组合 spring事务

问题引入

最近在开发一个自定义序列时 因为是服务器集群环境 所以使用到了分布式锁 来保证自定义序列的唯一性，在创建序列的方法中 使用 redisson来锁住创建序列的代码，并在该方法上加上了Spring的声明式事务，也就是@Transactional 注解。最后在使用jmetter模拟并发测试中，发现存在多条重复的序列，经过debug发现是因为 在我释放掉锁之后，spring才提交事务，这样其他的请求线程获取到锁就有可能查询到旧数据，导致产生重复序列。

@Transactional(rollbackFor = Exception.class)
public String getSequence(int type) {
    
  RedissonClient redissonClient = SpringContextUtis.getBean("redissonClient", RedissonClient.class);
  SequenceMapper sequenceMapper = SpringContextUtis.getBean("sequenceMapper", SequenceMapper.class);
  RLock lock = redissonClient.getLock("sequence");
  // 加锁
  lock.lock();
  System.err.println("get lock");
  try {
    
      // 执行查询操作
      // 执行修改操作
  } finally {
    
      System.err.println("abandon lock");
      // 解锁 
      // 在这里打断点 往下走就可以看到spring事务的提交或回滚流程
      // 也就是说 在这里我们释放掉锁之后 其实 事务并没有提交 具体可以查看源码
      lock.unlock();
}

解决方式

弄清楚产生问题的原因后其实解决方式就显而易见了。这里最简单的方式就是去掉 事务注解。如果不想去掉注解，可以参照以下实现方式。

第一种方式

既然是锁释放掉之后事务没有提交造成的，那我在锁释放之前提交事务就ok了，最简单的实现就是 在套一层方法，在外层的方法中锁住创建序列的代码，然后将事务中的锁移除掉就行

    public String lockUserSeq(){
    
        RedissonClient redissonClient = SpringContextUtis.getBean("redissonClient", RedissonClient.class);
        RLock lock = redissonClient.getLock("sequence");
        String value = null;
        while(value == null) {
    
            lock.lock();
            try {
    
            	// 这里的话为了保证 事务 不失效，采用自己注入自己的方式
            	// 也可以通过 SpringContextUtis工具获取对应的bean
               value = sequenceGenerate.getSeqNotLock(1);
            } finally {
    
                lock.unlock();
            }
        }
        return value;
    }

第二种方式

借助于mysql的事务隔离级别来实现，这里的话需要确认你的mysql是innerDB引擎，MyISAM 引擎是不支持事务的。

// 直接指定事务隔离级别 为串行化 就ok了 这样好像也没必要用redisson了...
@Transactional(rollbackFor = Exception.class, isolation = Isolation.SERIALIZABLE)

当然还有其他解决方式 这里就不再一一列举了，因为跑题了…

redisson 组合 spring 事务 解决

因为我们是服务器集群，所以必然会有很多这种场景，每次都这样写，太过繁琐，一不小心忘记了就是一个大事故了… 所以针对这种场景，我们需要一种简单的解决方式，毫无疑问，在方法上绑定注解就是最简单的。这种方式下，我们只需要对既需要分布式锁来保证并发安全问题，又需要事务管理的方法上绑定我们定义的注解就解决问题了，快速简单。

定义注解

定义的注解里面建议组合spring的Transactional注解 这样的话，使用者能够定义像传播特性、隔离级别等之类的属性，而不需要我们重新定义

/**
 * @author pzz
 * @Date 2023-06-14 9:29
 * @implNote
 */

@Target(ElementType.METHOD)
@Retention(RetentionPolicy.RUNTIME)
public @interface RedissonLock {
    

    // 开启事务 如果不需要组合的话 可选择使用此注解 
    // 但需要稍微改一下后续的代码
//    boolean transactional() default false;
	// 组合事务注解 最推荐 很方便
    Transactional transactional();

    // 开启锁
    boolean lock() default true;

    // 锁的名字
    String lockName();

    // 超时时间
    long timeOut() default 30;

    // 单位
    TimeUnit unit() default TimeUnit.SECONDS;
}

为注解添加处理器

当然，光有注解是不行的。我们需要一个对应的处理器，来对注解进行解析和功能的实现。这里参考 Spring事务的实现方式来实现， 本质上就是通过代理来实现的。简单来说(其实实现也很简单)我们可以在spring创建bean对象的时候判断bean中的方法是否存在@RedisLock注解，如果存在，那么我们为该类创建代理。

// 可以使用其他方式注入这个bean 不过这种方式是最简单的
// 注意 我们需要实现 BeanPostProcessor 
// 这样我们就可以扩展自己的bean创建时的处理逻辑
@Component
public final class RedissonLockManager implements BeanPostProcessor {
    
	// 给出两种方式来实现 可实现 before 或者 after 都行
//    @Override
//    public Object postProcessBeforeInitialization(Object bean, String beanName) throws BeansException {
    
//        // 循环判断bean的每个方式是否标记了 @RedissonLock 注解 如果有标记 则为他生成代理
//        Method[] methods = bean.getClass().getDeclaredMethods();
//        Object proxy;
//        for (Method method : methods) {
    
//            if (NullableBean.doEquals((proxy = interceptor(bean, method)))) {
    
//                continue;
//            }
//            return proxy;
//        }
//        return bean;
//    }

    @Override
    public Object postProcessAfterInitialization(Object bean, String beanName) throws BeansException {
    
        /**
         * 如果是代理对象 获取真正的被代理的类的类对象
         * 必须要这样写 如果实现after方法来实现的话 
         * 因为如果类中存在@Transaction 注解的话 这里传入的bean就是代						   		  理对象了 
         */
        Class<?> targetClass = AopUtils.getTargetClass(bean);
        // 循环判断bean的每个方式是否标记了 @RedissonLock 注解 如果有标记 则为他生成代理
        Method[] methods = targetClass.getDeclaredMethods();
        Object proxy;
        for (Method method : methods) {
    
            if (null == (proxy = interceptor(bean, method))) {
    
                continue;
            }
            return proxy;
        }
        return bean;
    }
	// 这里也是保持和 spring 事务一致 只允许公共方法
    public boolean isPublic(int mod) {
    
        return Modifier.isPublic(mod);
    }
    
	// 判断是否存在 @RedissonLock 注解
    public RedissonLock hasRedissonLock(Method method) {
    
        return method.getAnnotation(RedissonLock.class);
    }

    /**
     * 如果条件都满足 那么我们对这个类进行代理
     */
    public Object interceptor(Object bean, Method method) {
    
        RedissonLock redissonLock;
        if (isPublic(method.getModifiers()) && (redissonLock = hasRedissonLock(method)) != null) {
    
            // 进入这里 我们就代理这个类 ProxyUtils 会在下面给出来
            return ProxyUtils.getProxy(bean, this::methodInterceptor);
        }
        return null;
    }

	// 这里是具体的代理逻辑
    public Object methodInterceptor(MethodInvocation invocation) throws Throwable {
    
        Method method = invocation.getMethod();
        RedissonLock redissonLock = hasRedissonLock(method);
        // 如果调用的方法不存在 redissonLock 注解 直接返回
        // 确切的说是给下一个链去执行他的处理逻辑
        if (redissonLock != null) {
    
            RedissonClient redissonClient = SpringContextUtis.getBean("redissonClient", RedissonClient.class);
            // 这里加锁
            RLock lock = redissonClient.getLock(redissonLock.lockName());
            lock.lock(redissonLock.timeOut(), redissonLock.unit());
            try {
    
                // 判断是否开启事务
                Transactional transactional = redissonLock.transactional();
                if (transactional != null) {
    
                    // 这里是我们手动对spring事务进行管理
                    // 下面会给出这个类的具体实现
                    // 这里就是开启事务
                    TransactionStatus status = TransactionalManager.begin(transactional);
                    // 用来存储返回
                    Object proceed = null;
                    try {
    
                    	// 执行被代理的方法的逻辑
                         proceed = invocation.proceed();
                         // 测试回滚
                    // throw new RuntimeException();     
                    }catch (Exception e){
    
                    	// 如果方法出现异常 我们手动进行回退
                        e.printStackTrace();
                        TransactionalManager.rollback(status);
                    }
                    // 提交事务
                    TransactionalManager.commit(status);
                    return proceed;
                }
                return invocation.proceed();
            } finally {
    
            	// 这里对锁进行释放 保证了事务要么已经提交，要么已经回滚
            	// 不会存在 最开始的问题了
                lock.unlock();
            }
        }
        // 不存在直接交给下一个链执行
        return invocation.proceed();
    }
}

实现自己的事务管理器

其实本来这个类是作为一个静态内部类放入到 上面定义的 RedissonLockManager 中的 但那样代码太长了，不便于观看，所以单独移出来，我们只需要在 RedissonLockManager 进行静态导入就可以不用改其中的代码了。

public final class TransactionalManager {
    
		// 获取spring中的事务管理器
        private static PlatformTransactionManager getManager() {
    
            return SpringContextUtis.getBean(PlatformTransactionManager.class);
        }

		// 开启事务 这里是对不组合 @Transactional注解 的实现
		// 所有的 事务配置 都采用默认的
        public static TransactionStatus begin() {
    
            TransactionStatus status = getManager().getTransaction(TransactionDefinition.withDefaults());
            return status;
        }
		
		// 接收一个事务属性对象来开启事务
        public static TransactionStatus begin(TransactionAttribute attribute){
    
            TransactionStatus status = getManager().getTransaction(attribute);
            return status;
        }
		
		// 接收一个 @Transactional 注解对象来开启事务
        public static TransactionStatus begin(Transactional transactional){
    
        	// 这里是spring提供的事务对象解析器
        	// 可以将注解对象转换成事务属性对象 
            SpringTransactionAnnotationParser parser = new SpringTransactionAnnotationParser();
            return begin(parser.parseTransactionAnnotation(transactional));
        }

		// 提交事务
        public static void commit(TransactionStatus status) {
    
            getManager().commit(status);
        }

		// 关闭事务
        public static void rollback(TransactionStatus status) {
    
            getManager().rollback(status);
        }

    }

代理工具类

public final class ProxyUtils {
    

    private ProxyUtils(){
    }

    public static Object getProxy(Object target, org.aopalliance.intercept.MethodInterceptor methodInterceptor){
    
        ProxyFactory proxyFactory = new ProxyFactory();
        proxyFactory.setTarget(target);
        proxyFactory.addAdvice(methodInterceptor);
        return proxyFactory.getProxy();
    }
}

大功告成，开始测试

这里提供一段测试代码

controller

@RestController
public class UserController {
    

	@Resource
    private SequenceGenerate sequenceGenerate;
    @Resource
    private SequenceUniqueRepository sequenceUniqueRepository;

   @GetMapping("/guest/getSeq")
    public String getSeq(){
    
//        String userSeq = sequenceGenerate.getSeqNotLock(1);
        String userSeq = sequenceGenerate.getSeqNotLock(1);
        SequenceUnique sequenceUnique = new SequenceUnique();
        sequenceUnique.setSeq(userSeq);
        sequenceUniqueRepository.save(sequenceUnique);
        return userSeq;
    }
}

service 实现

   @RedissonLock(lockName = "sequence",transactional = @Transactional())
    public String getSeqNotLock(int type) {
    
        // 执行查询操作
        // 执行修改操作
        return sequence.getSeq();
    }

使用 jmeter 进行测试

查看数据库中是否存在重复的序列字符串

SELECT count(id) as c,seq from test_sequence_unique GROUP BY seq having c > 1

当然，肯定是不存在的，不然我也不会放出来了…

表中存在1000条记录 说明所有请求都执行成功了

可以看到 不存在一条重复的记录 说明注解已经生效了

结语

功能至此算是实现了，如果各位有发现实现中存在的一些我没考虑到的问题，欢迎在评论区中指出来。