细粒度-秒杀限流-落地实现

2020-08-11 本文已影响0人大黑跟小白的日常

需求分析

这里有10种商品参与秒杀活动，为了提升用户参与体验，如下要求：
每种商品参与秒杀的数量不一样，每种商品要求的秒杀限流速率不一样；
比如：
华为P30-8G-256G-红色对应在当前秒杀场次活动种，可供秒杀的数量为10台，限速 1台/s，10秒放完10台；
华为挂式耳机****型号对应在当前秒杀场次活动种，可供秒杀的数量为100副，限速 10副/s，10秒放完100台；
华为matebook笔记本对应在当前秒杀场次活动种，可供秒杀的数量为5台，限速 0.5台/s，10秒放完5台；
...

需求整明白，上代码

pom依赖

        <dependency>
             <groupId>com.google.guava</groupId>
             <artifactId>guava</artifactId>
             <version>28.0-jre</version>
        </dependency>

自定义注解类，用于aop增强

/**
 * @Author G_Y
 * @Date 2020/8/11 18:42
 * @Description: // 自定义切点注解
 **/
@Inherited
@Documented
@Target({ElementType.METHOD})
@Retention(RetentionPolicy.RUNTIME)
public @interface AccessLimit {
    String pName() default "";
}

增强类

/**
 * @Author G_Y
 * @Date 2020/8/11 18:43
 * @Description: // 增强类
 **/
@Component
@Aspect
public class AccessLimitAop implements InitializingBean {

    private Map<Long, RateLimiter> rateLimiterHashMap = new HashMap<>();

    {
        rateLimiterHashMap.put(-1L, RateLimiter.create(10));// 默认限流速率
    }

    @Override
    public void afterPropertiesSet() throws Exception {
        // 程序启动 或 利用 定时任务 或 利用 配置中心，将 秒杀商品对应的 限流速率 加载到程序
        Map<Long, Integer> goodsLimitRate = new HashMap<>();
        goodsLimitRate.put(111111L, 2); // 商品111111 对应 限流速率 为  一秒5个
        goodsLimitRate.put(111112L, 3);
        goodsLimitRate.put(111113L, 5);
        // 如上比如是 从数据库加载到的 商品 对应 的限流速率 map

        // 初始化rateLimiterHashMap
        goodsLimitRate.forEach((k, v) -> {
            rateLimiterHashMap.put(k, RateLimiter.create(v));
        });
    }

    @Pointcut("@annotation(com.changgou.web.order.aop.AccessLimit)")
    public void limit() {
    }


    @Around("limit()")
    public Object around(ProceedingJoinPoint proceedingJoinPoint) {

        // 获取当前访问的方法签名信息
        MethodSignature signature = (MethodSignature) proceedingJoinPoint.getSignature();
        // 获取当前执行方法的注解对象
        AccessLimit annotation = signature.getMethod().getAnnotation(AccessLimit.class);
        // 获取 注解属性
        String pName = annotation.pName();
        // pName = "#goodsId"

        RateLimiter rateLimiter = null;
        if (StringUtils.isEmpty(pName)) {
            rateLimiter = rateLimiterHashMap.get(-1L);
        } else {
            // 获取方法参数名称
            String[] parameterNames = signature.getParameterNames();
            // 有顺序的 ["userId","goodsId"]
            // 将方法参数名称跟对应的值 封装到 map
            Map<String, Object> paramMap = new HashMap<String, Object>();
            for (int i = 0; i < parameterNames.length; i++) {
                String parameterName = parameterNames[i];
                paramMap.put(parameterName, proceedingJoinPoint.getArgs()[i]);
            }
            if (!pName.startsWith("#")) {
                throw new RuntimeException("aop 参数 非法");
            }
            pName = pName.substring(1);//goodsId
            Object value = paramMap.get(pName);
            if (value == null)
                throw new RuntimeException("aop 参数 非法, 参数中未找到" + pName);
            if (!(value instanceof Long))
                throw new RuntimeException("aop 参数 非法, 参数类型需要为Long, 参数不满足条件 " + pName);
            Long goodsId = (Long) value;

            rateLimiter = rateLimiterHashMap.get(goodsId);

        }

        // 实现限流
        boolean flag = rateLimiter.tryAcquire();
        if (flag) {
            try {
                return proceedingJoinPoint.proceed();
            } catch (Throwable throwable) {
                throwable.printStackTrace();
                return new Result<>(false, 9000, "对不起,抢单失败了,请下次再来！");
            }
        } else {
            // 前端发现返回9000应该用js做页面转圈处理，抢单中，10秒中后再提示抢单结果，优化用户体验。
            return new Result<>(false, 9000, "对不起,抢单失败了,请下次再来！");
        }
    }

}

使用方式

/**
 * @Author G_Y
 * @Date 2020/8/11 19:21
 * @Description: // 测试限流接口
 **/
@RestController
public class TestController {
    // 秒杀，某 用户 抢 某个 商品，抢单成功 返回 订单id(写死，秒杀下单业务略)
    @AccessLimit(pName = "#goodsId")
    @RequestMapping("/test/limit")
    public Result<Long> testRateLimiteSecKill(Integer userId, Long goodsId) {
        return new Result(true, 2000, "抢单成功", 8888889999L);
    }

}

模拟客户端测试

/**
 * @Author G_Y
 * @Date 2020/8/11 19:28
 * @Description: // 测试秒杀抢单
 **/
public class AccessClientTest {
    ExecutorService fixedThreadPool = Executors.newFixedThreadPool(10);
    /**
     * get请求
     * @param realUrl
     * @return
     */
    public String sendGet(URL realUrl) {
        String result = "";
        BufferedReader in = null;
        try {
            // 打开和URL之间的连接
            URLConnection connection = realUrl.openConnection();
            // 设置通用的请求属性
            connection.setRequestProperty("accept", "*/*");
            connection.setRequestProperty("connection", "Keep-Alive");
            connection.setRequestProperty("user-agent",
                    "Mozilla/4.0 (compatible; MSIE 6.0; Windows NT 5.1;SV1)");
            // 建立实际的连接
            connection.connect();

            // 定义 BufferedReader输入流来读取URL的响应
            in = new BufferedReader(new InputStreamReader(
                    connection.getInputStream()));
            String line;
            while ((line = in.readLine()) != null) {
                result += line;
            }
        } catch (Exception e) {
            System.out.println("发送GET请求出现异常！" + e);
            e.printStackTrace();
        }
        // 使用finally块来关闭输入流
        finally {
            try {
                if (in != null) {
                    in.close();
                }
            } catch (Exception e2) {
                e2.printStackTrace();
            }
        }
        return result;
    }

    public void access() throws Exception{
        final URL url = new URL("http://localhost:9011/test/limit?userId=777&goodsId=111113");
        for(int i=0;i<10;i++) {
            fixedThreadPool.submit(new Runnable() {
                public void run() {
                    System.out.println(sendGet(url));
                }
            });
        }
        TimeUnit.SECONDS.sleep(1);
        System.out.println("--------------------------");
        for(int i=0;i<10;i++) {
            fixedThreadPool.submit(new Runnable() {
                public void run() {
                    System.out.println(sendGet(url));
                }
            });
        }
        fixedThreadPool.shutdown();
        fixedThreadPool.awaitTermination(Long.MAX_VALUE, TimeUnit.SECONDS);
    }
    public static void main(String[] args) throws Exception{
        AccessClientTest accessClient = new AccessClientTest();
        accessClient.access();
    }
}