分布式秒杀

通常在具体的业务中，平台方会发布秒杀席位个数，秒杀的时间段，让各个商家报名，将本身的产品参与秒杀活动。这里将同事画的一张图放上来，大体是这么一个流程。关于秒杀原理能够参考单机秒杀系统样例

这里面关于Nginx的限流能够参考高并发之Nginx的限流

咱们这里在秒杀的前一天将全部的批准参加秒杀的商家商品数据导入到redis的商品队列中，这里咱们使用当当的elastic-job来实现，关于elastic-job的使用，能够参考分布式调度Elastic-Job攻略

这里是以商家的服务来当成一件商品来处理的，因此service能够理解成商品。这里有一个区位的概念，即秒杀当天商家能够报名的全部时间段，咱们称之为区位，好比0点～2点，2点～4点等等。区位的实体类字段以下。

/**
 * 秒杀区位
 */
@Data
@AllArgsConstructor
@NoArgsConstructor
public class Location {
    private Long id; //区位id
    private LocalDate date; //日期
    private TimeSegment timeSegment; //时间段
    private int locationNumber; //区位数量，便可以容纳多少商家报名
    private BigDecimal baseCost; //商家参与活动的最低缴纳金
    private double sellPercent;  //销售百分比缴纳率
    private boolean releaseLocation; //是否发布
}

/**
 * 秒杀服务
 */
@RequiredArgsConstructor
@NoArgsConstructor
public class Service {
    @Getter
    @Setter
    @NonNull
    private Long id; //服务id
    @Getter
    @Setter
    @NonNull
    private Long signUpId; //报名id
    @Getter
    @Setter
    @NonNull
    private String serviceCode; //服务编码
    @Getter
    @Setter
    @NonNull
    private String serviceName; //服务名称
    @Getter
    @Setter
    @NonNull
    private String serviceLevel; //服务分类
    @Getter
    @Setter
    @NonNull
    private Price price; //价格，包含普通价格和秒杀价格
    @Getter
    @Setter
    @NonNull
    private int totalCount; //参与秒杀的总数
    @Getter
    @Setter
    @NonNull
    private int limitCount; //单个用户能够购买该服务的数量
    @Getter
    @Setter
    @NonNull
    private TimeSegment timeSegment; //时间段
    @Getter
    @Setter
    @NonNull
    private CheckStatus checkStatus; //平台对该秒杀的审核状态
    @Getter
    private Lock lock = new ReentrantLock(); //重入锁
    @Getter
    private Condition condition = lock.newCondition(); //重入锁条件
}

商家商品(服务)数据导入到redis的商品队列中

/**
 * 在天天的0点开始处理，查看后一天是否有秒杀活动
 */
@Slf4j
@Component
@ElasticSimpleJob(cron="0 0 0 * * ?",jobName="loadRedis",shardingTotalCount=2,jobParameter="日期",shardingItemParameters="0=Load0,1=Load1")
public class LoadDataToRedisJob implements SimpleJob {
    @Autowired
    private RedisService redisService;
    @Autowired
    private DataDao dataDao;

    @Override
    public void execute(ShardingContext shardingContext) {
        //获取后一天全部的秒杀区位
        List<Location> locationNextDayList = dataDao.findJobToday(LocalDate.now().plusDays(1));
        if (!CollectionUtils.isEmpty(locationNextDayList)) {
            //获取一天全部的秒杀区位时间段
            List<TimeSegment> segmentList = locationNextDayList.stream().map(location -> location.getTimeSegment())
                    .collect(Collectors.toList());
            switch (shardingContext.getShardingItem()) {
                case 0:
                    //获取每一个时间段内的全部参与秒杀的服务
                    segmentList.stream().map(timeSegment -> {
                        List<Service> serviceInSegment = dataDao.findServiceInSegment(timeSegment);
                        serviceInSegment.stream().forEach(service -> service.setTimeSegment(timeSegment));
                        return serviceInSegment;
                        //扁平化全部服务，统一为一组List
                    }).flatMap(services -> services.stream()).forEach(service -> {
                        //以服务id以及秒杀时间段组合为主键
                        String key = service.getId() + service.getTimeSegment().toString();
                        //若是redis中存在该主键的队列，则清空队列
                        if (redisService.exists(key)) {
                            for (int i = 0; i < redisService.llen(key); i++) {
                                redisService.rpop(key);
                            }
                        }
                        //清空后，根据每一个服务的参与总数，将服务按总数量推送到该主键队列中
                        for (int i = 0; i < service.getTotalCount(); i++) {
                            redisService.lpush(key, JSONObject.toJSONString(service));
                        }
                        log.info(service.getId() + service.getTimeSegment().toString());
                        //以服务id+":count"组合成该服务的总数键，若是redis中存在该键，则删除
                        String countKey = service.getId() + ":count";
                        if (redisService.exists(countKey)) {
                            redisService.del(countKey);
                        }
                        //从新将总数放入该键的redis中存储
                        redisService.set(countKey, String.valueOf(service.getTotalCount()));
                    });
                    break;
                case 1:
                    break;
                default:
                    break;
            }
        }

    }
}

将服务导入到redis队列后，咱们须要设立一个秒杀活动开始的标识，让秒杀下单只能在秒杀活动进行中开启，不在秒杀时间内不容许下单。

/**
 * 给秒杀时间点设立开启标识，天天0点开始，每2小时执行一次
 */
@Slf4j
@Component
@ElasticSimpleJob(cron="0 0 0/2 * * ?",jobName="openTimeSeg",shardingTotalCount=1,jobParameter="日期",shardingItemParameters="0=Open0")
public class OpenTimeSegmentJob implements SimpleJob {
    @Autowired
    private RedisService redisService;
    @Autowired
    private DataDao dataDao;

    @Override
    public void execute(ShardingContext shardingContext) {
        //获取当天的全部秒杀区位
        List<Location> locationToDayList = dataDao.findJobToday(LocalDate.now());
        //若是当天有秒杀活动
        if (!CollectionUtils.isEmpty(locationToDayList)) {
            //获取当前时间点，当前时间点不必定是准点
            LocalDateTime now = LocalDateTime.now();
            int year = now.getYear();
            int month = now.getMonthValue();
            int day = now.getDayOfMonth();
            int hour = now.getHour();
            //将当前时间拼装成整点
            LocalDateTime beginDate = LocalDateTime.of(year, month, day, hour, 0, 0);
            //以整点时间为基准，在redis中放入开启秒杀时间段，119分钟后消失（每一个时间段段为1小时59分钟，2小时的最后一分钟结束该时间段秒杀）
            redisService.set("TimeStart:" + new TimeSegment(beginDate, beginDate.plusMinutes(119)).toString(),
                    "opened",7140);
            log.info(beginDate.toString() + "至" + beginDate.plusMinutes(119).toString() + "秒杀开始");
        }
    }
}

到了秒杀时间，咱们就能够开始下单了，先定义一个秒杀单的接口

public interface SecOrder {
    /**
     * 秒杀下单
     * @param secOrder
     * @return
     */
    public String makeOrder(SecOrder secOrder);

    /**
     * 是否存在该订单编号的秒杀单
     * @param orderNo
     * @return
     */
    public boolean exitsOrder(String orderNo);

    /**
     * 修改支付状态
     * @param orderNo
     */
    public void changePayStatus(String orderNo);
}

SecOrder的实现类的各属性以下。

/**
 * 服务秒杀单
 */
@Slf4j
@AllArgsConstructor
@NoArgsConstructor
@ServiceSecOrderVersion(value = 1)
public class ServiceSecOrder implements SecOrder {
    @Getter
    @Setter
    private Long id; //订单id
    @Getter
    @Setter
    private String orderNo; //订单编号
    @Getter
    @Setter
    private Service service; //订单服务内容
    @Getter
    @Setter
    private TimeSegment timeSegment; //秒杀时间段
    @Getter
    @Setter
    private int number; //订单数量
    @Getter
    @Setter
    private BigDecimal amount; //订单金额
    @Getter
    @Setter
    private AppUser user; //下单人
    @Getter
    @Setter
    private int orderStatus; //订单状态
    @Getter
    @Setter
    private LocalDateTime createDate; //建立日期

而后开始下秒杀订单

@Override
public String makeOrder(SecOrder secOrder) {
    RedisService redisService = SpringBootUtil.getBean(RedisService.class);
    IdService idService = SpringBootUtil.getBean(IdService.class);
    MessageSender messageSender = SpringBootUtil.getBean(MessageSender.class);
    //若是当前时间在秒杀时间段以外，不容许秒杀下单
    if (LocalDateTime.now().isBefore(((ServiceSecOrder)secOrder).getTimeSegment().getBeginTime())
    || LocalDateTime.now().isAfter(((ServiceSecOrder)secOrder).getTimeSegment().getEndTime())) {
        throw new RuntimeException("不在秒杀时间段内");
    }
    //因为测试时间的问题，此处须要屏蔽，等到实际部署时须要恢复
    LocalDateTime now = LocalDateTime.now();
    int year = now.getYear();
    int month = now.getMonthValue();
    int day = now.getDayOfMonth();
    int hour = now.getHour();
    LocalDateTime beginDate = LocalDateTime.of(year,month,day,hour,0,0);
    //从redis中检查是否有开启秒杀时间段
    if (!redisService.exists("TimeStart:" + new TimeSegment(beginDate,beginDate.plusMinutes(119)).toString())) {
        throw new RuntimeException("当前时间段无秒杀");
    }
    ((ServiceSecOrder)secOrder).setId(idService.genId());
    if (((ServiceSecOrder)secOrder).getNumber() > ((ServiceSecOrder)secOrder).getService().getLimitCount()) {
        throw new RuntimeException("秒杀数量超出限购");
    }
    AppUser loginAppUser = AppUserUtil.getLoginAppUser();
    AppUser user = new AppUser();
    user.setId(loginAppUser.getId());
    ((ServiceSecOrder)secOrder).setOrderNo(getCode(idService));
    user.setUsername(loginAppUser.getUsername());
    ((ServiceSecOrder)secOrder).setUser(user);
    //设置订单状态0表示未支付状态
    ((ServiceSecOrder)secOrder).setOrderStatus(0);
    ((ServiceSecOrder)secOrder).setCreateDate(LocalDateTime.now());
    //设置用户秒杀队列键名(每一种服务都有独立的用户秒杀队列)
    //队列名由User:+服务id+时间段组成
    String key = "User:" + ((ServiceSecOrder)secOrder).getService().getId()
            + ((ServiceSecOrder)secOrder).getTimeSegment().toString();
    String serviceKey = ((ServiceSecOrder)secOrder).getService().getId()
            + ((ServiceSecOrder)secOrder).getService().getTimeSegment().toString();
    //若是服务队列还有数据，则推送用户进队列，不然直接返回秒杀失败
    if (redisService.llen(serviceKey) > 0) {
        //创建每一个服务对每一个用户对限制数量对键,格式为——"服务id:用户id"
        String limitUserKey = ((ServiceSecOrder) secOrder).getService().getId() + ":"
                + ((ServiceSecOrder) secOrder).getUser().getId();
        //若是该键存在，获取该键的值(这里须要考虑分布式的并发问题的可能)
        //可是有幂等，因此此处不会出现一个用户同时秒杀一个商品(服务)的多个并发线程存在
        if (redisService.exists(limitUserKey)) {
            String limitCount = redisService.get(limitUserKey);
            //若是该键的值达到服务商品的限制数，返回秒杀失败
            if (Integer.valueOf(limitCount) == ((ServiceSecOrder) secOrder).getService().getLimitCount()) {
                return ((ServiceSecOrder) secOrder).getUser().getUsername() + "秒杀服务"
                        + ((ServiceSecOrder) secOrder).getService().getServiceName() + "失败";
            }
        }else {  //若是不存在，设置该键的值为0
            redisService.set(limitUserKey,"0");
        }
        //将秒杀用户id推送到该队列中
        redisService.lpush(key, ((ServiceSecOrder) secOrder).getUser().getId() + "");

咱们将用户id推送到redis队列后就要开始匹配秒杀结果了，由于商品队列早已经在前一天就推送进去了。

/**
 * 秒杀结果匹配任务，天天0点开始，每2小时执行一次
 */
@Slf4j
@Component
@RabbitListener(queues = {SecendKillMq.SECENDKILL_QUEUE + "_1",
                        SecendKillMq.SECENDKILL_QUEUE + "_2",
                        SecendKillMq.SECENDKILL_QUEUE + "_3",
                        SecendKillMq.SECENDKILL_QUEUE + "_4",
                        SecendKillMq.SECENDKILL_QUEUE + "_5",
                        SecendKillMq.SECENDKILL_QUEUE + "_6",
                        SecendKillMq.SECENDKILL_QUEUE + "_7",
                        SecendKillMq.SECENDKILL_QUEUE + "_8",
                        SecendKillMq.SECENDKILL_QUEUE + "_9",
                        SecendKillMq.SECENDKILL_QUEUE + "_10"})
@ElasticSimpleJob(cron="0 0 0/2 * * ?",jobName="secResult",shardingTotalCount=2,jobParameter="日期",shardingItemParameters="0=SecKill0,1=SecKill1")
public class SecendKillResultJob implements SimpleJob {
    @Autowired
    private RedisService redisService;
    @Autowired
    private DataDao dataDao;
    //为了跟MQ搭配，唤醒中断用
    private List<Service> serviceList;

    @Override
    public void execute(ShardingContext shardingContext) {
        //获取当天全部的秒杀区位
        List<Location> locationTodayList = dataDao.findJobToday(LocalDate.now());
        //若是当天有秒杀活动
        if (!CollectionUtils.isEmpty(locationTodayList)) {
            //获取一天全部的秒杀区位时间段
            List<TimeSegment> segmentList = locationTodayList.stream().map(location -> location.getTimeSegment())
                    .collect(Collectors.toList());
            switch (shardingContext.getShardingItem()) {
                case 0:
                    //从全部秒杀区位时间段过滤当前秒杀时间段
                    this.serviceList = segmentList.stream().filter(timeSegment -> LocalDateTime.now().isAfter(timeSegment.getBeginTime()) &&
                            LocalDateTime.now().isBefore(timeSegment.getEndTime()))
                            //将时间段转化成时间段内的秒杀服务
                            .map(timeSegment -> {
                                List<Service> serviceInSegment = dataDao.findServiceInSegment(timeSegment);
                                serviceInSegment.stream().forEach(service -> service.setTimeSegment(timeSegment));
                                return serviceInSegment;
                                //扁平化全部的秒杀服务，将全部当前时间段内的服务放入serviceList属性中
                                //就是拿出当前时间段内全部参与秒杀的服务
                            }).flatMap(services -> services.stream()).collect(Collectors.toList());
                    //并行化处理全部的秒杀服务
                    int lism = 0;
                    if (serviceList.size() > Runtime.getRuntime().availableProcessors() * 2) {
                        lism = serviceList.size();
                    }else {
                        lism = Runtime.getRuntime().availableProcessors() * 2;
                    }
                    ForkJoinPool forkJoinPool = new ForkJoinPool(lism);
                    try {
                        forkJoinPool.submit(() ->
                        this.serviceList.parallelStream().forEach(service -> {
                            while (true) {
                                try {
                                    service.getLock().lock();
                                    String userKey = "User:" + service.getId() + service.getTimeSegment().toString();
                                    String serviceKey = service.getId() + service.getTimeSegment().toString();
                                    String countKey = service.getId() + ":count";
                                    //若是下秒杀时间内没有用户下单该服务，则中断该服务的并行线程
                                    //若是有用户下单则唤醒该并行线程
                                    while (redisService.llen(userKey).equals(0L)) {
                                        try {
                                            log.info("用户队列无数据，开始中断");
                                            service.getCondition().await();
                                            LocalDateTime now = LocalDateTime.now();
                                            if (now.isAfter(service.getTimeSegment().getEndTime())) {
                                                break;
                                            }
                                        } catch (InterruptedException e) {
                                            e.printStackTrace();
                                        }
                                    }

到这里，若是没有用户下单，则会进行线程中断，不会去执行while (true)的无限循环。SecendKillResultJob同时又是RabbitMQ的一个消费者，同时监听了10个消息队列，监听后进行以下处理

@RabbitHandler
public void receice(byte[] data, Channel channel, Message message) throws IOException {
    try {
        //告诉服务器收到这条消息 已经被我消费了 能够在队列删掉；不然消息服务器觉得这条消息没处理掉 后续还会在发
        channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);
        //收到服务id
        Long serviceId = unSerialize(data);
        log.info(serviceId + "");
        //若是当前秒杀服务列表不为空
        if (!CollectionUtils.isEmpty(this.serviceList)) {
            //从服务列表中过滤出id为MQ收取的服务ID的服务
            this.serviceList.stream().filter(service -> service.getId().equals(serviceId))
                    .forEach(service -> {
                        log.info("存在" + service.getId());
                        try {
                            service.getLock().lock();
                            //对该服务所在线程进行唤醒
                            service.getCondition().signalAll();
                        } finally {
                            service.getLock().unlock();
                        }
                    });
        }
    } catch (IOException e) {
        e.printStackTrace();
        //丢弃这条消息
        channel.basicNack(message.getMessageProperties().getDeliveryTag(), false,false);
        log.info("receiver fail");
    }
}

/**
 * 反序列化
 * @param data
 * @return
 */
private Long unSerialize(byte[] data) {
    Input input = null;
    try {
        Kryo kryo = new Kryo();
        input = new Input(new ByteArrayInputStream(data));
        return kryo.readObject(input,Long.class);
    }
    finally {
        input.close();
    }
}

这样咱们再回到ServiceSecOrder的makeOrder下单方法中，将用户下单的服务id异步发送到MQ中，去唤醒秒杀结果匹配任务继续执行。

/**
 * 服务秒杀单
 */
@Slf4j
@AllArgsConstructor
@NoArgsConstructor
@ServiceSecOrderVersion(value = 1)
public class ServiceSecOrder implements SecOrder {
    @Getter
    @Setter
    private Long id; //订单id
    @Getter
    @Setter
    private String orderNo; //订单编号
    @Getter
    @Setter
    private Service service; //订单服务内容
    @Getter
    @Setter
    private TimeSegment timeSegment; //秒杀时间段
    @Getter
    @Setter
    private int number; //订单数量
    @Getter
    @Setter
    private BigDecimal amount; //订单金额
    @Getter
    @Setter
    private AppUser user; //下单人
    @Getter
    @Setter
    private int orderStatus; //订单状态
    @Getter
    @Setter
    private LocalDateTime createDate; //建立日期

    @Override
    public String makeOrder(SecOrder secOrder) {
        RedisService redisService = SpringBootUtil.getBean(RedisService.class);
        IdService idService = SpringBootUtil.getBean(IdService.class);
        MessageSender messageSender = SpringBootUtil.getBean(MessageSender.class);
        //若是当前时间在秒杀时间段以外，不容许秒杀下单
        if (LocalDateTime.now().isBefore(((ServiceSecOrder)secOrder).getTimeSegment().getBeginTime())
        || LocalDateTime.now().isAfter(((ServiceSecOrder)secOrder).getTimeSegment().getEndTime())) {
            throw new RuntimeException("不在秒杀时间段内");
        }
        //因为测试时间的问题，此处须要屏蔽，等到实际部署时须要恢复
        LocalDateTime now = LocalDateTime.now();
        int year = now.getYear();
        int month = now.getMonthValue();
        int day = now.getDayOfMonth();
        int hour = now.getHour();
        LocalDateTime beginDate = LocalDateTime.of(year,month,day,hour,0,0);
        //从redis中检查是否有开启秒杀时间段
        if (!redisService.exists("TimeStart:" + new TimeSegment(beginDate,beginDate.plusMinutes(119)).toString())) {
            throw new RuntimeException("当前时间段无秒杀");
        }
        ((ServiceSecOrder)secOrder).setId(idService.genId());
        if (((ServiceSecOrder)secOrder).getNumber() > ((ServiceSecOrder)secOrder).getService().getLimitCount()) {
            throw new RuntimeException("秒杀数量超出限购");
        }
        AppUser loginAppUser = AppUserUtil.getLoginAppUser();
        AppUser user = new AppUser();
        user.setId(loginAppUser.getId());
        ((ServiceSecOrder)secOrder).setOrderNo(getCode(idService));
        user.setUsername(loginAppUser.getUsername());
        ((ServiceSecOrder)secOrder).setUser(user);
        //设置订单状态0表示未支付状态
        ((ServiceSecOrder)secOrder).setOrderStatus(0);
        ((ServiceSecOrder)secOrder).setCreateDate(LocalDateTime.now());
        //设置用户秒杀队列键名(每一种服务都有独立的用户秒杀队列)
        //队列名由User:+服务id+时间段组成
        String key = "User:" + ((ServiceSecOrder)secOrder).getService().getId()
                + ((ServiceSecOrder)secOrder).getTimeSegment().toString();
        String serviceKey = ((ServiceSecOrder)secOrder).getService().getId()
                + ((ServiceSecOrder)secOrder).getService().getTimeSegment().toString();
        //若是服务队列还有数据，则推送用户进队列，不然直接返回秒杀失败
        if (redisService.llen(serviceKey) > 0) {
            //创建每一个服务对每一个用户对限制数量对键,格式为——"服务id:用户id"
            String limitUserKey = ((ServiceSecOrder) secOrder).getService().getId() + ":"
                    + ((ServiceSecOrder) secOrder).getUser().getId();
            //若是该键存在，获取该键的值(这里须要考虑分布式的并发问题的可能)
            //可是有幂等，因此此处不会出现一个用户同时秒杀一个商品(服务)的多个并发线程存在
            if (redisService.exists(limitUserKey)) {
                String limitCount = redisService.get(limitUserKey);
                //若是该键的值达到服务商品的限制数，返回秒杀失败
                if (Integer.valueOf(limitCount) == ((ServiceSecOrder) secOrder).getService().getLimitCount()) {
                    return ((ServiceSecOrder) secOrder).getUser().getUsername() + "秒杀服务"
                            + ((ServiceSecOrder) secOrder).getService().getServiceName() + "失败";
                }
            }else {  //若是不存在，设置该键的值为0
                redisService.set(limitUserKey,"0");
            }
            //将秒杀用户id推送到该队列中
            redisService.lpush(key, ((ServiceSecOrder) secOrder).getUser().getId() + "");
            //唤醒秒杀结果匹配任务继续执行。
            CompletableFuture.runAsync(() -> {
                messageSender.send(SecendKillMq.MQ_EXCHANGE_SECENDKILL,
                        SecendKillMq.ROUTING_KEY_SECENDKILL,
                        ((ServiceSecOrder) secOrder).getService().getId());
            });

这里咱们须要先看一段redis-lua代码，表示若是用户队列，服务队列均有数据，则将其弹出，放入一个hash中做为匹配的结果，同时扣减服务队列总数。其意义能够参考用户金额的终极解决方案--Redis Lua

@Override
public Object secondKill(String userKey, String serviceKey,String userResult, String countKey,String serviceId) {
    String script = "if redis.call('llen',KEYS[1]) > 0 and redis.call('llen',KEYS[2]) > 0 " +
            "and tonumber(redis.call('get',KEYS[4])) > 0 then " +
            "local userid = redis.call('rpop',KEYS[1])" +
            "redis.call('hset',KEYS[3],userid,redis.call('rpop',KEYS[2])) " +
            "redis.call('decr',KEYS[4]) " +
            "return redis.call('incr',KEYS[5]..':'..userid) else return 0 end";
    return execute(jedis -> jedis.eval(script,5,userKey,serviceKey,userResult,countKey,serviceId));
}

秒杀服务线程唤醒后，继续执行

/**
 * 秒杀结果匹配任务，天天0点开始，每2小时执行一次
 */
@Slf4j
@Component
@RabbitListener(queues = {SecendKillMq.SECENDKILL_QUEUE + "_1",
                        SecendKillMq.SECENDKILL_QUEUE + "_2",
                        SecendKillMq.SECENDKILL_QUEUE + "_3",
                        SecendKillMq.SECENDKILL_QUEUE + "_4",
                        SecendKillMq.SECENDKILL_QUEUE + "_5",
                        SecendKillMq.SECENDKILL_QUEUE + "_6",
                        SecendKillMq.SECENDKILL_QUEUE + "_7",
                        SecendKillMq.SECENDKILL_QUEUE + "_8",
                        SecendKillMq.SECENDKILL_QUEUE + "_9",
                        SecendKillMq.SECENDKILL_QUEUE + "_10"})
@ElasticSimpleJob(cron="0 0 0/2 * * ?",jobName="secResult",shardingTotalCount=2,jobParameter="日期",shardingItemParameters="0=SecKill0,1=SecKill1")
public class SecendKillResultJob implements SimpleJob {
    @Autowired
    private RedisService redisService;
    @Autowired
    private DataDao dataDao;
    //为了跟MQ搭配，唤醒中断用
    private List<Service> serviceList;

    @Override
    public void execute(ShardingContext shardingContext) {
        //获取当天全部的秒杀区位
        List<Location> locationTodayList = dataDao.findJobToday(LocalDate.now());
        //若是当天有秒杀活动
        if (!CollectionUtils.isEmpty(locationTodayList)) {
            //获取一天全部的秒杀区位时间段
            List<TimeSegment> segmentList = locationTodayList.stream().map(location -> location.getTimeSegment())
                    .collect(Collectors.toList());
            switch (shardingContext.getShardingItem()) {
                case 0:
                    //从全部秒杀区位时间段过滤当前秒杀时间段
                    this.serviceList = segmentList.stream().filter(timeSegment -> LocalDateTime.now().isAfter(timeSegment.getBeginTime()) &&
                            LocalDateTime.now().isBefore(timeSegment.getEndTime()))
                            //将时间段转化成时间段内的秒杀服务
                            .map(timeSegment -> {
                                List<Service> serviceInSegment = dataDao.findServiceInSegment(timeSegment);
                                serviceInSegment.stream().forEach(service -> service.setTimeSegment(timeSegment));
                                return serviceInSegment;
                                //扁平化全部的秒杀服务，将全部当前时间段内的服务放入serviceList属性中
                                //就是拿出当前时间段内全部参与秒杀的服务
                            }).flatMap(services -> services.stream()).collect(Collectors.toList());
                    //并行化处理全部的秒杀服务
                    int lism = 0;
                    if (serviceList.size() > Runtime.getRuntime().availableProcessors() * 2) {
                        lism = serviceList.size();
                    }else {
                        lism = Runtime.getRuntime().availableProcessors() * 2;
                    }
                    ForkJoinPool forkJoinPool = new ForkJoinPool(lism);
                    try {
                        forkJoinPool.submit(() ->
                        this.serviceList.parallelStream().forEach(service -> {
                            while (true) {
                                try {
                                    service.getLock().lock();
                                    String userKey = "User:" + service.getId() + service.getTimeSegment().toString();
                                    String serviceKey = service.getId() + service.getTimeSegment().toString();
                                    String countKey = service.getId() + ":count";
                                    //若是下秒杀时间内没有用户下单该服务，则中断该服务的并行线程
                                    //若是有用户下单则唤醒该并行线程
                                    while (redisService.llen(userKey).equals(0L)) {
                                        try {
                                            log.info("用户队列无数据，开始中断");
                                            service.getCondition().await();
                                            LocalDateTime now = LocalDateTime.now();
                                            if (now.isAfter(service.getTimeSegment().getEndTime())) {
                                                break;
                                            }
                                        } catch (InterruptedException e) {
                                            e.printStackTrace();
                                        }
                                    }
                                    log.info("中断被唤醒，继续运行");
                                    //若是用户队列和服务队列均有数据
                                    while (redisService.llen(userKey) > 0 && redisService.llen(serviceKey) > 0) {
                                        //匹配出秒杀结果，并扣减服务数量，增长用户秒杀过该服务的数量
                                        redisService.secondKill(userKey, serviceKey, "UserResult" + service.getId(), countKey,String.valueOf(service.getId()));
                                    }
                                    //若是服务队列为空，表示被秒杀完了，从用户队列弹出用户，告知秒杀失败
                                    while (redisService.llen(serviceKey).equals(0L)) {
                                        redisService.hset("UserResult" + service.getId(),redisService.rpop(userKey),"秒杀失败");
                                        if (redisService.llen(userKey).equals(0L)) {
                                            break;
                                        }
                                    }
                                    //当前时间已经超出了秒杀时间段，结束while(true)无限循环
                                    LocalDateTime now = LocalDateTime.now();
                                    if (now.isAfter(service.getTimeSegment().getEndTime())) {
                                        break;
                                    }
                                } finally {
                                    service.getLock().unlock();
                                }
                            }
                        })).get();
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    } catch (ExecutionException e) {
                        e.printStackTrace();
                    }
                    break;
                case 1:
                    break;
                default:
                    break;
            }
        }
    }

当有用户秒杀到服务时，或者服务被秒杀完，用户的下单须要知道本身是否秒杀成功或者秒杀失败。

/**
 * 服务秒杀单
 */
@Slf4j
@AllArgsConstructor
@NoArgsConstructor
@ServiceSecOrderVersion(value = 1)
public class ServiceSecOrder implements SecOrder {
    @Getter
    @Setter
    private Long id; //订单id
    @Getter
    @Setter
    private String orderNo; //订单编号
    @Getter
    @Setter
    private Service service; //订单服务内容
    @Getter
    @Setter
    private TimeSegment timeSegment; //秒杀时间段
    @Getter
    @Setter
    private int number; //订单数量
    @Getter
    @Setter
    private BigDecimal amount; //订单金额
    @Getter
    @Setter
    private AppUser user; //下单人
    @Getter
    @Setter
    private int orderStatus; //订单状态
    @Getter
    @Setter
    private LocalDateTime createDate; //建立日期

    @Override
    public String makeOrder(SecOrder secOrder) {
        RedisService redisService = SpringBootUtil.getBean(RedisService.class);
        IdService idService = SpringBootUtil.getBean(IdService.class);
        MessageSender messageSender = SpringBootUtil.getBean(MessageSender.class);
        //若是当前时间在秒杀时间段以外，不容许秒杀下单
        if (LocalDateTime.now().isBefore(((ServiceSecOrder)secOrder).getTimeSegment().getBeginTime())
        || LocalDateTime.now().isAfter(((ServiceSecOrder)secOrder).getTimeSegment().getEndTime())) {
            throw new RuntimeException("不在秒杀时间段内");
        }
        //因为测试时间的问题，此处须要屏蔽，等到实际部署时须要恢复
        LocalDateTime now = LocalDateTime.now();
        int year = now.getYear();
        int month = now.getMonthValue();
        int day = now.getDayOfMonth();
        int hour = now.getHour();
        LocalDateTime beginDate = LocalDateTime.of(year,month,day,hour,0,0);
        //从redis中检查是否有开启秒杀时间段
        if (!redisService.exists("TimeStart:" + new TimeSegment(beginDate,beginDate.plusMinutes(119)).toString())) {
            throw new RuntimeException("当前时间段无秒杀");
        }
        ((ServiceSecOrder)secOrder).setId(idService.genId());
        if (((ServiceSecOrder)secOrder).getNumber() > ((ServiceSecOrder)secOrder).getService().getLimitCount()) {
            throw new RuntimeException("秒杀数量超出限购");
        }
        AppUser loginAppUser = AppUserUtil.getLoginAppUser();
        AppUser user = new AppUser();
        user.setId(loginAppUser.getId());
        ((ServiceSecOrder)secOrder).setOrderNo(getCode(idService));
        user.setUsername(loginAppUser.getUsername());
        ((ServiceSecOrder)secOrder).setUser(user);
        //设置订单状态0表示未支付状态
        ((ServiceSecOrder)secOrder).setOrderStatus(0);
        ((ServiceSecOrder)secOrder).setCreateDate(LocalDateTime.now());
        //设置用户秒杀队列键名(每一种服务都有独立的用户秒杀队列)
        //队列名由User:+服务id+时间段组成
        String key = "User:" + ((ServiceSecOrder)secOrder).getService().getId()
                + ((ServiceSecOrder)secOrder).getTimeSegment().toString();
        String serviceKey = ((ServiceSecOrder)secOrder).getService().getId()
                + ((ServiceSecOrder)secOrder).getService().getTimeSegment().toString();
        //若是服务队列还有数据，则推送用户进队列，不然直接返回秒杀失败
        if (redisService.llen(serviceKey) > 0) {
            //创建每一个服务对每一个用户对限制数量对键,格式为——"服务id:用户id"
            String limitUserKey = ((ServiceSecOrder) secOrder).getService().getId() + ":"
                    + ((ServiceSecOrder) secOrder).getUser().getId();
            //若是该键存在，获取该键的值(这里须要考虑分布式的并发问题的可能)
            //可是有幂等，因此此处不会出现一个用户同时秒杀一个商品(服务)的多个并发线程存在
            if (redisService.exists(limitUserKey)) {
                String limitCount = redisService.get(limitUserKey);
                //若是该键的值达到服务商品的限制数，返回秒杀失败
                if (Integer.valueOf(limitCount) == ((ServiceSecOrder) secOrder).getService().getLimitCount()) {
                    return ((ServiceSecOrder) secOrder).getUser().getUsername() + "秒杀服务"
                            + ((ServiceSecOrder) secOrder).getService().getServiceName() + "失败";
                }
            }else {  //若是不存在，设置该键的值为0
                redisService.set(limitUserKey,"0");
            }
            //将秒杀用户id推送到该队列中
            redisService.lpush(key, ((ServiceSecOrder) secOrder).getUser().getId() + "");
            //唤醒秒杀结果匹配任务继续执行。
            CompletableFuture.runAsync(() -> {
                messageSender.send(SecendKillMq.MQ_EXCHANGE_SECENDKILL,
                        SecendKillMq.ROUTING_KEY_SECENDKILL,
                        ((ServiceSecOrder) secOrder).getService().getId());
            });
            LocalDateTime start = LocalDateTime.now();
            //从redis的匹配结果获取当前用户的秒杀结果
            Future<String> future = CompletableFuture.supplyAsync(() -> {
                while (true) {
                    if (redisService.hexists("UserResult" + ((ServiceSecOrder) secOrder).getService().getId(),
                            ((ServiceSecOrder) secOrder).getUser().getId() + "")) {
                        return redisService.hget("UserResult" + ((ServiceSecOrder) secOrder).getService().getId(),
                                ((ServiceSecOrder) secOrder).getUser().getId() + "");
                    }
                    if (LocalDateTime.now().isAfter(start.plusSeconds(3))) {
                        return "秒杀失败";
                    }
                }
            });
            try {
                if (future.get(3000, TimeUnit.MILLISECONDS).equals("秒杀失败")) {
                    log.info(((ServiceSecOrder) secOrder).getUser().getUsername() + "秒杀服务"
                            + ((ServiceSecOrder) secOrder).getService().getServiceName() + "失败");
                    return ((ServiceSecOrder) secOrder).getUser().getUsername() + "秒杀服务"
                            + ((ServiceSecOrder) secOrder).getService().getServiceName() + "失败";
                } else {
                    log.info(((ServiceSecOrder) secOrder).getUser().getUsername() + "秒杀服务"
                            + ((ServiceSecOrder) secOrder).getService().getServiceName() + "成功");
                    SecOrderDao secOrderDao = SpringBootUtil.getBean(SecOrderDao.class);
                    //秒杀成功的将秒杀单存入数据库
                    secOrderDao.saveServiceSecOrder((ServiceSecOrder) secOrder);
                    //等待支付结果，5分钟后根据支付与否进行处理
                    //若是未支付，则将被秒杀到的服务从新入服务队列，并增长服务总数
                    ServicePayBackSender servicePayBackSender = SpringBootUtil.getBean(ServicePayBackSender.class);
                    CompletableFuture.runAsync(() -> {
                        servicePayBackSender.send(SecendKillMq.MQ_EXCHANGE_DEAD,SecendKillMq.ROURING_KEY_DEAD,(ServiceSecOrder) secOrder);
                    });
                    return ((ServiceSecOrder) secOrder).getUser().getUsername() + "秒杀服务"
                            + ((ServiceSecOrder) secOrder).getService().getServiceName() + "成功";
                }
            } catch (Exception e) {
                e.printStackTrace();
                log.info(((ServiceSecOrder) secOrder).getUser().getUsername() + "秒杀服务"
                        + ((ServiceSecOrder) secOrder).getService().getServiceName() + "失败");
                return ((ServiceSecOrder) secOrder).getUser().getUsername() + "秒杀服务"
                        + ((ServiceSecOrder) secOrder).getService().getServiceName() + "失败";
            }
        } else {
            return ((ServiceSecOrder) secOrder).getUser().getUsername() + "秒杀服务"
                    + ((ServiceSecOrder) secOrder).getService().getServiceName() + "失败";
        }
    }

这里使用了RabbitMQ的延迟队列，配置以下

/**
 * rabbitmq配置
 *
 */
@Configuration
public class RabbitmqConfig {

   @Bean
   public List<Queue> secendKillQueues() {
      List<Queue> queues = new ArrayList<>();
      for (int i = 1;i < 11;i++) {
         Queue queue = new Queue(SecendKillMq.SECENDKILL_QUEUE + "_" + i);
         queues.add(queue);
      }
      return queues;
   }

   @Bean
   public TopicExchange secendKillExchange() {
      return new TopicExchange(SecendKillMq.MQ_EXCHANGE_SECENDKILL);
   }
// @Bean
// public FanoutExchange secondKillExchange() {
//    return new FanoutExchange(SecendKillMq.MQ_EXCHANGE_SECENDKILL);
// }

   @Bean
   public List<Binding> bingingSecondKill(){
      List<Binding> bindings = new ArrayList<>();
      for (int i = 1;i < 11;i++) {
         Binding binding = BindingBuilder.bind(secendKillQueues().get(i - 1)).to(secendKillExchange())
               .with(SecendKillMq.ROUTING_KEY_SECENDKILL + "_" + i);
         bindings.add(binding);
      }
      return bindings;
   }

   @Bean
   public Queue deadQueue() {
      Map<String,Object> arguments = new HashMap<>();
      arguments.put("x-dead-letter-exchange",SecendKillMq.MQ_EXCHANGE_DEAD);
      arguments.put("x-dead-letter-routing-key",SecendKillMq.ROUTING_KEY_PAYBACK);
      return new Queue(SecendKillMq.DEAD_QUEUE,true,false,false,arguments);
   }

   @Bean
   public DirectExchange deadExchange() {
      return new DirectExchange(SecendKillMq.MQ_EXCHANGE_DEAD);
   }

   @Bean
   public Binding bindingDeadExchange() {
      return BindingBuilder.bind(deadQueue()).to(deadExchange())
            .with(SecendKillMq.ROURING_KEY_DEAD);
   }

   @Bean
   public Queue payBackQueue() {
      return new Queue(SecendKillMq.PAYBACK_QUEUE,true,false,false);
   }

   @Bean
   public Binding bindingPayBack() {
      return BindingBuilder.bind(payBackQueue()).to(deadExchange())
            .with(SecendKillMq.ROUTING_KEY_PAYBACK);
   }

}

消息生产者ServicePayBackSender以下

@Slf4j
@Component
public class ServicePayBackSender implements RabbitTemplate.ConfirmCallback,RabbitTemplate.ReturnCallback {
    @Autowired
    private RabbitTemplate rabbitTemplate;

    public void send(String exchange,String routingKey,Object content) {
        log.info("send content=" + content);
        this.rabbitTemplate.setMandatory(true);
        this.rabbitTemplate.setConfirmCallback(this);
        this.rabbitTemplate.setReturnCallback(this);
        MessagePostProcessor processor = message -> {
            message.getMessageProperties().setExpiration(300000 + "");
            return message;
        };
        this.rabbitTemplate.convertAndSend(exchange,routingKey,serialize(content),processor);
    }

    /**
     * 确认后回调:
     * @param correlationData
     * @param ack
     * @param cause
     */
    @Override
    public void confirm(CorrelationData correlationData, boolean ack, String cause) {
        if (!ack) {
            log.info("send ack fail, cause = " + cause);
        } else {
            log.info("send ack success");
        }
    }

    /**
     * 失败后return回调：
     *
     * @param message
     * @param replyCode
     * @param replyText
     * @param exchange
     * @param routingKey
     */
    @Override
    public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {
        log.info("send fail return-message = " + new String(message.getBody()) + ", replyCode: " + replyCode + ", replyText: " + replyText + ", exchange: " + exchange + ", routingKey: " + routingKey);
    }

    /**
     * 对消息对象进行二进制序列化
     * @param o
     * @return
     */
    private byte[] serialize(Object o) {
        Kryo kryo = new Kryo();
        ByteArrayOutputStream stream = new ByteArrayOutputStream();
        Output output = new Output(stream);
        kryo.writeObject(output, o);
        output.close();
        return stream.toByteArray();
    }
}

消费者以下

@Slf4j
@Component
@RabbitListener(queues = SecendKillMq.PAYBACK_QUEUE)
public class ServicePayBackDeal {
    @Autowired
    private SecOrderDao secOrderDao;
    @Autowired
    private RedisService redisService;

    @RabbitHandler
    public void receice(byte[] data, Channel channel, Message message) throws IOException {
        try {
            //告诉服务器收到这条消息 已经被我消费了 能够在队列删掉；不然消息服务器觉得这条消息没处理掉 后续还会在发
            channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);
            ServiceSecOrder order = unSerialize(data);
            if (secOrderDao.countServiceSecOrderHasPay(order) == 0) {
                String key = order.getService().getId() + order.getService().getTimeSegment().toString();
                String countKey = order.getService().getId() + ":count";
                String countLimit = order.getService().getId() + ":" + order.getUser().getId();
                redisService.unPayedBack(key, JSONObject.toJSONString(order.getService()),countKey,countLimit);
            }
        } catch (IOException e) {
            e.printStackTrace();
            //丢弃这条消息
            channel.basicNack(message.getMessageProperties().getDeliveryTag(), false,false);
            log.info("receiver fail");
        }
    }

    /**
     * 反序列化
     * @param data
     * @return
     */
    private ServiceSecOrder unSerialize(byte[] data) {
        Input input = null;
        try {
            Kryo kryo = new Kryo();
            input = new Input(new ByteArrayInputStream(data));
            return kryo.readObject(input,ServiceSecOrder.class);
        }
        finally {
            input.close();
        }
    }
}

unPayedBack也是一段redis-lua,代码以下

@Override
public Object unPayedBack(String serviceKey,String serviceValue, String countKey,String limitCountKey) {
    String script = "redis.call('lpush',KEYS[1],ARGV[1]) " +
            "redis.call('incr',KEYS[2]) " +
            "return redis.call('decr',KEYS[3])";
    return execute(jedis -> jedis.eval(script,3,serviceKey,countKey,limitCountKey,serviceValue));
}

在秒杀结束后唤醒全部的秒杀中断，退出while(true)的无限循环

/**
 * 在每一个秒杀段最后一分钟唤醒全部秒杀中断，天天的1点59分开始，每2小时执行一次
 */
@Component
@ElasticSimpleJob(cron="0 59 1/2 * * ?",jobName="signal",shardingTotalCount=2,jobParameter="日期",shardingItemParameters="0=Signal0,1=Signal1")
public class SignalJob implements SimpleJob {
    @Autowired
    private DataDao dataDao;
    @Autowired
    private MessageSender messageSender;

    @Override
    public void execute(ShardingContext shardingContext) {
        //获取当天全部的秒杀区位
        List<Location> locationTodayList = dataDao.findJobToday(LocalDate.now());
        if (!CollectionUtils.isEmpty(locationTodayList)) {
            //获取一天全部的秒杀区位时间段
            List<TimeSegment> segmentList = locationTodayList.stream().map(location -> location.getTimeSegment())
                    .collect(Collectors.toList());
            switch (shardingContext.getShardingItem()) {
                case 0:
                    segmentList.stream().filter(timeSegment -> LocalDateTime.now().plusMinutes(-2L).isAfter(timeSegment.getBeginTime()) &&
                            LocalDateTime.now().plusMinutes(-2L).isBefore(timeSegment.getEndTime()))
                            //将时间段转化成时间段内的秒杀服务
                            .map(timeSegment -> {
                                List<Service> serviceInSegment = dataDao.findServiceInSegment(timeSegment);
                                serviceInSegment.stream().forEach(service -> service.setTimeSegment(timeSegment));
                                return serviceInSegment;
                                //扁平化全部的秒杀服务
                            }).flatMap(services -> services.stream()).forEach(service -> {
                                CompletableFuture.runAsync(() -> {
                                    messageSender.send(SecendKillMq.MQ_EXCHANGE_SECENDKILL,
                                    SecendKillMq.ROUTING_KEY_SECENDKILL,
                                    service.getId());
                                });
                            });
                    break;
                case 1:
                    break;
                default:
                    break;
            }
        }
    }
}

如今咱们来创建下单的Controller

先创建一个秒杀单工厂接口

public interface SecOrderFactory {
    /**
     * 建立秒杀订单
     * @return
     */
    public SecOrder createSecOrder();

    /**
     * 获取秒杀订单
     * @return
     */
    public SecOrder getSecOrder();
}

Service版本标签

@Target(ElementType.TYPE)
@Retention(RetentionPolicy.RUNTIME)
public @interface ServiceSecOrderVersion {
    int value();
}

Service工厂实现类

@Component
public class ServiceSecOrderFactory implements SecOrderFactory {
    private Set<Class<?>> classes = ClassUtil.getClassSet("com.cloud.secondkill.domain");
    private SecOrder createdSecOrder;

    @PostConstruct
    private void init() {
        this.createdSecOrder = createSecOrder();
    }

    @Override
    public SecOrder createSecOrder() {
        Object instance = null;
        try {
            //过滤有@ServiceSecOrderVersion标签的类
            instance = classes.stream().filter(clazz -> clazz.isAnnotationPresent(ServiceSecOrderVersion.class))
                    //过滤实现了SecOrder接口的类
                    .filter(clazz -> SecOrder.class.isAssignableFrom(clazz))
                    //找出版本号大的类，并实例化为对象
                    .max(Comparator.comparingInt(clazz -> clazz.getAnnotation(ServiceSecOrderVersion.class).value()))
                    .get().newInstance();
        } catch (InstantiationException e) {
            e.printStackTrace();
        } catch (IllegalAccessException e) {
            e.printStackTrace();
        }
        return (SecOrder) instance;
    }

    @Override
    public SecOrder getSecOrder() {
        return createdSecOrder;
    }
}

秒杀单的Bean

@Component
public class SecOrderBean {
    @Getter
    private Map<String,Class> secOrderFactoryMap = new HashMap<>();
    @Getter
    private Map<String,Class> secOrderMap = new HashMap<>();

    @PostConstruct
    private void init() {
        Set<Class<?>> classes = ClassUtil.getClassSet("com.cloud.secondkill.domain");
        classes.stream().filter(clazz -> SecOrderFactory.class.isAssignableFrom(clazz))
                .forEach(clazz -> secOrderFactoryMap.put(clazz.getSimpleName(),clazz));
        classes.stream().filter(clazz -> SecOrder.class.isAssignableFrom(clazz))
                .forEach(clazz -> secOrderMap.put(clazz.getSimpleName(),clazz));
    }
}

下单Controller

@Slf4j
@RestController
public class SecOrderController {
    private ThreadLocal<SecOrderFactory>  secOrderFactory = new ThreadLocal<>();
    private ThreadLocal<SecOrder> secOrderService = new ThreadLocal<>();
    @Autowired
    private SecOrderBean secOrderBean;

    @SuppressWarnings("unchecked")
    @Transactional
    @PostMapping("/makesecorder")
    @LxRateLimit(perSecond = 500,timeOut = 500) //此处为接口限流，数字能够根据实际清空改写
    public Result<String> makeSecOrder(@RequestBody String secOrderStr, @RequestParam("type") String type) throws Exception {
        log.info(secOrderStr);
        try {
            SecOrder secOrder = setSecOrderFactory(secOrderStr, type);
            String secResult = this.secOrderService.get().makeOrder(secOrder);
            return Result.success(secResult);
        } finally {
            secOrderFactory.remove();
            secOrderService.remove();
        }
    }

    private SecOrder setSecOrderFactory(String secOrderStr,String type) {
        Class classType = secOrderBean.getSecOrderMap().get(type);
        Object secOrder = JSONObject.parseObject(secOrderStr, classType);
        setSecOrderFactory(type);
        return (SecOrder) secOrder;
    }

    private void setSecOrderFactory(String type) {
        Class classFactoryType = secOrderBean.getSecOrderFactoryMap().get(type + "Factory");
        this.secOrderFactory.set((SecOrderFactory) SpringBootUtil.getBean(classFactoryType));
        this.secOrderService.set(this.secOrderFactory.get().getSecOrder());
    }
}

以上不明白的能够参考用工厂方法模式来下不一样订单 以及使用简单工厂加接口加适配器模式来遵照开闭原则

@LxRateLimit为接口限流

@Target(ElementType.METHOD)
@Retention(RetentionPolicy.RUNTIME)
public @interface LxRateLimit {
    /**
     *
     * @return
     */
    String value() default "";

    /**
     * 每秒向桶中放入令牌的数量   默认最大即不作限流
     * @return
     */
    double perSecond() default Double.MAX_VALUE;

    /**
     * 获取令牌的等待时间  默认0
     * @return
     */
    int timeOut() default 0;

    /**
     * 超时时间单位
     * @return
     */
    TimeUnit timeOutUnit() default TimeUnit.MILLISECONDS;
}

AOP拦截

@Slf4j
@Aspect
@Component
public class LxRateLimitAspect {
    private RateLimiter rateLimiter = RateLimiter.create(Double.MAX_VALUE);

    /**
     * 带有指定注解切入
     */
    @ResponseBody
    @Around(value = "@annotation(com.cloud.secondkill.annotion.LxRateLimit)")
    public Object aroundNotice(ProceedingJoinPoint pjp) throws Throwable {
        log.info("拦截到了{}方法...", pjp.getSignature().getName());
        Signature signature = pjp.getSignature();
        MethodSignature methodSignature = (MethodSignature)signature;
        //获取目标方法
        Method targetMethod = methodSignature.getMethod();
        if (targetMethod.isAnnotationPresent(LxRateLimit.class)) {
            //获取目标方法的@LxRateLimit注解
            LxRateLimit lxRateLimit = targetMethod.getAnnotation(LxRateLimit.class);
            rateLimiter.setRate(lxRateLimit.perSecond());
            if (!rateLimiter.tryAcquire(lxRateLimit.timeOut(), lxRateLimit.timeOutUnit()))
                return "服务器繁忙，请稍后再试!";
        }
        return pjp.proceed();
    }
}

具体原理能够参考Guava RateLimiter限流源码解析和实例应用

如今要加入秒杀的幂等，来防止工具秒杀

token Controller

@RestController
public class TokenController {
    @Autowired
    private RedisService redisService;

    @GetMapping("/gettoken")
    public Map getToken(@RequestParam("url") String url) {
        Map<String,String> tokenMap = new HashMap<>();
        String tokenValue = UUID.randomUUID().toString();
        AppUser user = AppUserUtil.getLoginAppUser();
        String key = url + user.getId();
        tokenMap.put(key,tokenValue);
        redisService.set(key,tokenValue);
        return tokenMap;
    }
}

配置幂等Spring MVC拦截器

@Component
@Slf4j
public class TokenInterceptor implements HandlerInterceptor {
    @Autowired
    private RedisService redisService;

    @Override
    public boolean preHandle(HttpServletRequest request, HttpServletResponse response, Object handler) throws Exception {
        AppUser user = AppUserUtil.getLoginAppUser();
        String tokenName = request.getRequestURI() + user.getId();
        String tokenValue = request.getParameter("token_value");
        if (tokenValue != null && !tokenValue.equals("")) {
            log.info("tokenName:{},tokenValue:{}",tokenName,tokenValue);
            return handleToken(request,response,handler,user);
        }
        return false;
    }

    @Override
    public void postHandle(HttpServletRequest request, HttpServletResponse response, Object handler, @Nullable ModelAndView modelAndView) throws Exception {
        if (redisService.exists(request.getParameter("token_value"))) {
            RedisTool.releaseDistributedLock(redisService, request.getParameter("token_value"), request.getParameter("token_value"));
        }
    }

    @Override
    public void afterCompletion(HttpServletRequest request, HttpServletResponse response, Object handler, @Nullable Exception ex) throws Exception {

    }

    /**
     * 分布式锁处理
     * @param request
     * @param response
     * @param handler
     * @return
     * @throws Exception
     */
    private boolean handleToken(HttpServletRequest request, HttpServletResponse response, Object handler,AppUser user) throws Exception {
        //当大量高并发下全部带token参数的请求进来时，进行分布式锁定,容许某一台服务器的一个线程进入，锁定时间3分钟
        if (RedisTool.tryGetDistributedLock(redisService,request.getParameter("token_value"),request.getParameter("token_value"),180)) {
            if (redisService.exists(request.getRequestURI() + user.getId())) {
                //当请求的url与token与redis中的存储相同时
                if (redisService.get(request.getRequestURI() + user.getId()).equals(request.getParameter("token_value"))) {
                    //放行的该线程删除redis中存储的token
                    redisService.del(request.getRequestURI() + user.getId());
                    log.info("放行");
                    RedisTool.releaseDistributedLock(redisService,request.getParameter("token_value"),request.getParameter("token_value"));
                    //放行
                    return true;
                }
            }
            log.info("拦截");
            //当请求的url与token与redis中的存储不相同时，解除锁定
            RedisTool.releaseDistributedLock(redisService,request.getParameter("token_value"),request.getParameter("token_value"));
            //进行拦截
            return false;
        }
        return false;
    }
}

@SpringBootConfiguration
public class TokenInterceptorConfig extends WebMvcConfigurerAdapter {
    @Autowired
    private TokenInterceptor tokenInterceptor;

    @Override
    public void addInterceptors(InterceptorRegistry registry) {
        registry.addInterceptor(tokenInterceptor).addPathPatterns("/makesecorder");
}
}