SpringAOP及分布式锁示例

作者: 缓慢移动的蜗牛 | 来源:发表于2019-11-29 17:02 被阅读0次

SpringAOP及分布式锁示例
Redis分布式锁
锁（2）-- 分布式锁
六、AOP实现自动的系统日志功能
分布式锁的技术选型及思考
百度Java架构师分享分布式锁的技术选型及思考
阿里Java架构师分享Redis实现分布式锁的技术选型及思考
画分布式锁之"通文馆圣主"Curator的&
分布式锁
redis

AOP

AOP 的全称为 Aspect Oriented Programming，译为面向切面编程。实际上 AOP 就是通过预编译和运行期动态代理实现程序功能的统一维护的一种技术。在不同的技术栈中 AOP 有着不同的实现，但是其作用都相差不远，我们通过 AOP 为既有的程序定义一个切入点，然后在切入点前后插入不同的执行内容，以达到在不修改原有代码业务逻辑的前提下统一处理一些内容（比如日志处理、分布式锁）的目的。

为什么要使用 AOP

在实际的开发过程中，我们的应用程序会被分为很多层。通常来讲一个 Java 的 Web 程序会拥有以下几个层次：

Web 层：主要是暴露一些 Restful API 供前端调用。
业务层：主要是处理具体的业务逻辑。
数据持久层：主要负责数据库的相关操作（增删改查）。

虽然看起来每一层都做着全然不同的事情，但是实际上总会有一些类似的代码，比如日志打印和安全验证等等相关的代码。如果我们选择在每一层都独立编写这部分代码，那么久而久之代码将变的很难维护。所以我们提供了另外的一种解决方案: AOP。这样可以保证这些通用的代码被聚合在一起维护，而且我们可以灵活的选择何处需要使用这些代码。

AOP 的核心概念

切面（Aspect）：通常是一个类，在里面可以定义切入点和通知。
连接点（Joint Point）：被拦截到的点，因为 Spring 只支持方法类型的连接点，所以在 Spring 中连接点指的就是被拦截的到的方法，实际上连接点还可以是字段或者构造器。
切入点（Pointcut）：对连接点进行拦截的定义。
通知(有的地方叫增强)(Advice)： 拦截到连接点之后所要执行的代码，通知分为前置、后置、异常、最终、环绕通知五类。
目标(target)：被通知的对象。也就是需要加入额外代码的对象，也就是真正的业务逻辑被组织织入切面。
织入(Weaving)：把切面加入程序代码的过程。切面在指定的连接点被织入到目标对象中，在目标对象的生命周期里有多个点可以进行织入：

编译期：切面在目标类编译时被织入，这种方式需要特殊的编译器
类加载期：切面在目标类加载到JVM时被织入，这种方式需要特殊的类加载器，它可以在目标类被引入应用之前增强该目标类的字节码
运行期：切面在应用运行的某个时刻被织入，一般情况下，在织入切面时，AOP容器会为目标对象动态创建一个代理对象，Spring AOP就是以这种方式织入切面的。
AOP 代理：AOP 框架创建的对象，代理就是目标对象的加强。Spring 中的 AOP 代理可以使 JDK 动态代理，也可以是 CGLIB 代理，前者基于接口，后者基于子类。

Spring AOP

Spring 中的 AOP 代理还是离不开 Spring 的 IOC 容器，代理的生成，管理及其依赖关系都是由 IOC 容器负责，Spring 默认使用 JDK 动态代理，在需要代理类而不是代理接口的时候，Spring 会自动切换为使用 CGLIB 代理，不过现在的项目都是面向接口编程，所以 JDK 动态代理相对来说用的还是多一些。

Spring AOP相关注解

@Aspect：将一个java类定义为切面

@Pointcut：定义一个切入点，可以是一个规则表达式，比如某个 package 下的所有函数，也可以是一个注解等。例如：@Pointcut("execution(public * com.nanc.modules.*.service.imp..*.*(..))")或@annotation(myLog)

@Before(前置通知)： 在切入点开始处切入内容。

@After(后置通知)：在切入点结尾处切入内容。

@AfterReturning(最终通知)：在切入点 return 内容之后切入内容（可以用来对处理返回值做一些加工处理）。

@Around(环绕通知)：在切入点前后切入内容，并自己控制何时执行切入点自身的内容。

@AfterThrowing(异常通知)：用来处理当切入内容部分抛出异常之后的处理逻辑。

5类通知的执行顺序

@Around > @Before > @Around > @After > @AfterReturning

AOP的执行顺序

在实际情况下，我们对同一个接口做多个切面，比如日志打印、分布式锁、权限校验等等。这时候我们就会面临一个优先级的问题，这么多的切面该如何告知 Spring 执行顺序呢？这就需要我们定义每个切面的优先级，我们可以使用 @Order(i) 注解来标识切面的优先级, i 的值越小，优先级越高。假设现在我们一共有两个切面，一个 WebLogAspect，我们为其设置 @Order(100)；而另外一个切面 DistributeLockAspect 设置为 @Order(99)，所以 DistributeLockAspect 有更高的优先级，这个时候执行顺序是这样的：在 @Before 中优先执行 @Order(99) 的内容，再执行 @Order(100) 的内容。而在 @After 和 @AfterReturning 中则优先执行 @Order(100) 的内容，再执行 @Order(99) 的内容，可以理解为先进后出的原则。

案例演示

创建SpringBoot项目，使用的是Intelli IDEA 工具

配置

server:
  port: 8888
  servlet:
    context-path: /

spring:
  redis:
    host: 127.0.0.1

记录日志/统计方法执行用时

在实际的开发过程中，我们会需要将接口的出请求参数、返回数据甚至接口的消耗时间都以日志的形式打印出来以便排查问题，有些比较重要的接口甚至还需要将这些信息写入到数据库。而这部分代码相对来讲比较相似，为了提高代码的复用率，我们可以以 AOP 的方式将这种类似的代码封装起来。

//日志注解
@Retention(RetentionPolicy.RUNTIME)
@Target({ElementType.METHOD})
public @interface MyWebLog {
    String value() default "日志信息";
}

//切面类
@Slf4j
@Component
@Aspect
@Order(100)
public class WebLogAspect {

    /**
     * ThreadLocal 配合@After和@@AfterReturning 统计执行用时
     */
    private static ThreadLocal<Map<String, Object>> threadLocal = new ThreadLocal<>();

    private static final String START_TIME = "startTime";
    private static final String REQUEST_PARAMS = "requestParams";

    /**
     * 切点表达式
     *
     * 切点表达式：
     * execution 代表所要执行的表达式主体
     * 第一处 * 代表方法返回类型 *代表所有类型
     * 第二处 {@code com.nanc.*.controller} 包名代表aop监控的类所在的包
     * 第三处 .. 代表该包以及其子包下的所有类
     * 第四处 * 代表类名，*代表所有类
     * 第五处 *(..) *代表类中的方法名，(..)表示方法中的任何参数
     */
    @Pointcut("execution(public * com.nanc.*.controller..*.*(..))")
    public void pointcut(){
    }


    /**
     * 前置通知
     * @param joinPoint
     */
    @Before(value = "pointcut()")
    public void before(JoinPoint joinPoint){
        //log.info("@Before方法：{}.{} 运行，参数列表是：{}", joinPoint.getTarget().getClass().getCanonicalName(),joinPoint.getSignature().getName(), Arrays.asList(args));
        long startTime = System.currentTimeMillis();

        Map<String, Object> threadInfo = new HashMap<>();
        threadInfo.put(START_TIME, startTime);
        // 请求参数。
        StringBuilder requestStr = new StringBuilder();
        Object[] args = joinPoint.getArgs();
        if (args != null && args.length > 0) {
            for (Object arg : args) {
                requestStr.append(arg.toString());
            }
        }
        threadInfo.put(REQUEST_PARAMS, requestStr.toString());

        threadLocal.set(threadInfo);
        log.info("{}方法开始调用:requestData={}", joinPoint.getSignature().getName(), threadInfo.get(REQUEST_PARAMS));
    }

    /**
     * 最终通知
     *
     * JoinPoint一定要出现在参数表的第一位
     * @param joinPoint
     * @param result
     */
    @AfterReturning(value="pointcut()",returning="result")
    public void afterReturn(JoinPoint joinPoint,Object result){
        Map<String, Object> threadInfo = threadLocal.get();
        long takeTime = System.currentTimeMillis() - (long) threadInfo.getOrDefault(START_TIME, System.currentTimeMillis());

        threadLocal.remove();
        log.info("{}方法结束调用:耗时={}ms,result={}", joinPoint.getSignature().getName(), takeTime, result);
    }


    /**
     * 后置通知
     * 符合切点表达的方法 或者 有@MyWebLog的方法
     * @param joinPoint
     */
    @After(value = "pointcut()")
    public void after(JoinPoint joinPoint){
        Object[] args = joinPoint.getArgs();
        log.info("@After 方法：{}.{} 运行，参数列表是：{}", joinPoint.getTarget().getClass().getCanonicalName(),joinPoint.getSignature().getName(), Arrays.asList(args));
    }

    /**
     * 环绕通知
     * 统计@MyWebLog注解方法的运行时间
     * @param joinPoint
     */
    @Around(value = "@annotation(myWebLog))")
    public Object around(ProceedingJoinPoint joinPoint, MyWebLog myWebLog) throws Throwable{
        //统计一下方法的运行时间
        long start = System.currentTimeMillis();
        // 执行目标 service
        Object result = joinPoint.proceed();
        log.info("@Around 方法 {} 运行的时间为：{}", joinPoint.getSignature().getName(), (System.currentTimeMillis()-start));

        return result;
    }


    /**
     * 异常通知
     * @param joinPoint
     * @param exception
     */
    @AfterThrowing(value="pointcut()",throwing="exception")
    public void logException(JoinPoint joinPoint,Exception exception){
        log.info("@AfterThrowing 方法：{} 异常。。。异常信息：{}", joinPoint.getSignature().getName(), exception);
    }
}

分布式锁

我们程序中多多少少会有一些共享的资源或者数据，在某些时候我们需要保证同一时间只能有一个线程访问或者操作它们。在传统的单机部署的情况下，我们简单的使用 Java 提供的并发相关的 API 处理即可。但是现在大多数服务都采用分布式的部署方式，我们就需要提供一个跨进程的互斥机制来控制共享资源的访问，这种互斥机制就是我们所说的分布式锁。

注意

互斥性。在任时刻，只有一个客户端能持有锁。

不会发生死锁。即使有一个客户端在持有锁的期间崩溃而没有主动解锁，也能保证后续其他客户端能加锁。这个其实只要我们给锁加上超时时间即可。

具有容错性。只要大部分的 Redis 节点正常运行，客户端就可以加锁和解锁。

解铃还须系铃人。加锁和解锁必须是同一个客户端，客户端自己不能把别人加的锁给解了。

//切面类注解
@Retention(RetentionPolicy.RUNTIME)
@Target({ElementType.METHOD})
public @interface DistributeLock {
    /**
     * 分布式锁的 key 值
     * @return
     */
    String key();
    /**
     * 锁的超时时间
     * @return
     */
    long timeout() default 5;
    /**
     * 时间单位，默认秒
     * @return
     */
    TimeUnit timeUnit() default TimeUnit.SECONDS;
}


//切面类
@Slf4j
@Aspect
@Component
@Order(99)
public class DistributeLockAspect {

    @Autowired
    private StringRedisTemplate stringRedisTemplate;
    @Autowired
    private AnnotationResolver annotationResolver;

    @Pointcut("execution(* com.nanc.myaop.controller..*.*(..))")
    public void pointcut() {}

    /**
     * 指定表达式 并且 有@DistributeLock注解的
     * @param joinPoint
     * @param distributeLock
     * @return
     * @throws Exception
     */
    @Around(value = "pointcut() && @annotation(distributeLock)")
    public Object around(ProceedingJoinPoint joinPoint, DistributeLock distributeLock) throws Exception {
        String key = annotationResolver.resolver(joinPoint, distributeLock.key());
        String keyValue = getLock(key, distributeLock.timeout(), distributeLock.timeUnit());
        if (StringUtils.isEmpty(keyValue)) {
            //获取锁失败
            return BaseResponse.addError(ErrorCodeEnum.OPERATE_FAILED, "请勿频繁操作");
        }

        try{
            return joinPoint.proceed();
        }catch (Throwable e){
            return BaseResponse.addError(ErrorCodeEnum.SYSTEM_ERROR, "系统异常");
        }finally {
            unLock(key, keyValue);
        }
    }


    /**
     * 获取锁
     * @param key 锁的key
     * @param timeout 锁超时时间
     * @param timeUnit 时间单位
     *
     * @return
     */
    private String getLock(String key, long timeout, TimeUnit timeUnit) {
        try {
            String value = UUID.randomUUID().toString();
            Boolean lockStat = stringRedisTemplate.execute((RedisCallback<Boolean>)
                    connection -> connection
                            .set(
                                    key.getBytes(Charset.forName("UTF-8")),
                                    value.getBytes(Charset.forName("UTF-8")),
                                    Expiration.from(timeout, timeUnit),
                                    RedisStringCommands.SetOption.SET_IF_ABSENT
                            )
            );
            if (!lockStat) {
                return null;
            }
            return value;
        }catch (Exception e){
            log.error("获取分布式锁失败", e);
            return null;
        }
    }

    /**
     * 释放锁
     * @param key 锁的key
     * @param value 获取锁的时候存入的值
     */
    private void unLock(String key, String value){
        try{
            String script = "if redis.call('get', KEYS[1]) == ARGV[1] then return redis.call('del', KEYS[1]) else return 0 end";
            Boolean unLockStat = stringRedisTemplate.execute((RedisCallback<Boolean>)
                    connection -> connection.eval(
                            script.getBytes(), ReturnType.BOOLEAN, 1,
                            key.getBytes(Charset.forName("UTF-8")), value.getBytes(Charset.forName("UTF-8"))
                    )
            );
            if (!unLockStat) {
                log.error("释放分布式锁失败，key={}，已自动超时，其他线程可以重新获取锁", key);
            }
        }catch (Exception e){
            log.error("释放锁失败", e);
        }
    }
}

控制层代码

@RestController
public class TestController {

    @Autowired
    private TestService testService;

    @GetMapping("/hello/{name}")
    public String sayHello(@PathVariable(value = "name")String name) throws Exception{
        return testService.hello(name);
    }


    @PostMapping("/post-test")
    @DistributeLock(key = "post_test_#{baseRequest.channel}", timeout = 10)
    public BaseResponse postTest(@RequestBody @Valid BaseRequest baseRequest) {
        try {
            Thread.sleep(10000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        return BaseResponse.addResult();
    }
}

execution() 表达式解析

示例execution(* com.nanc.myaop.controller..*.*(..))

标识符	含义
execution（）	表达式的主体
第一个 * 符号	表示返回值的类型，* 代表所有返回类型
com.nanc.myaop.controller	AOP 所切的服务的包名，即需要进行横切的业务类
包名后面的 ..	表示当前包及子包
第二个 *	表示类名，* 表示所有类
最后的 .*(..)	第一个 . 表示任何方法名，括号内为参数类型，.. 代表任何类型参数

SpringAOP及分布式锁示例
AOP AOP 的全称为 Aspect Oriented Programming，译为面向切面编程。实际上 AOP...
Redis分布式锁
Redis的常用场景 [TOC] ★ Redis分布式锁示例代码, 其实该分布式锁的实现是存在很多问题.此处仅为...
锁（2）-- 分布式锁
前言：锁分3种：java锁、分布式锁、DB锁分布式锁的几种实现方式目前几乎很多大型网站及应用都是分布式部署...
六、AOP实现自动的系统日志功能
一、本课目标掌握SpringAOP的配置二、使用SpringAOP实现日志输出在下面的这个示例中，通过Spr...
分布式锁的技术选型及思考
本文来自作者一行在 GitChat 分享的{分布式锁的技术选型及思考} 锁和分布式锁在计算机中，锁的作用是解...
百度Java架构师分享分布式锁的技术选型及思考
本文来自作者一行在 GitChat 分享的{分布式锁的技术选型及思考} 锁和分布式锁在计算机中，锁的作用是解...
阿里Java架构师分享Redis实现分布式锁的技术选型及思考
本文来自作者一行在 GitChat 分享的{分布式锁的技术选型及思考} 锁和分布式锁在计算机中，锁的作用是解...
画分布式锁之"通文馆圣主"Curator的&
上一篇，我们基于示例和源码去剖析了可重入互斥锁，不可重入互斥锁，信号量锁，参见【画分布式锁之"通文馆圣主"C...
分布式锁
为什么要用分布式锁数据库乐观锁redis分布式锁zookeeper分布式锁使用分布式锁的场景实现分布式锁的方...
redis
1.redis 分布式锁及问题 (1)实现：加锁：setnx、解锁：del、锁超时：expire (2)可能出现...