@Configuration
public class WebServerConfig implements WebServerFactoryCustomizer<ConfigurableWebServerFactory> {
    @Override
    public void customize(ConfigurableWebServerFactory factory) {
        // 获取tomcat连接器
        ((TomcatServletWebServerFactory) factory).addConnectorCustomizers((TomcatConnectorCustomizer) connector -> {
            // 获取protocol
            Http11NioProtocol protocolHandler = (Http11NioProtocol) connector.getProtocolHandler();
            // 如果keepalive连接30s，还没有人使用，释放此链接
            protocolHandler.setKeepAliveTimeout(30000);
            // 允许开启最大长连接数量
            protocolHandler.setMaxKeepAliveRequests(10000);
        });
    }
}

@Configuration
public class GuavaCacheConfig {
    /**
     * 定义一个GuavaCache对象
     */
    private Cache<String, Object> guavaCache = null;

    @PostConstruct
    public void init() {
        guavaCache = CacheBuilder.newBuilder()
                .initialCapacity(10)
                .maximumSize(100)
                // 设置缓存写入后过期时间
                .expireAfterWrite(60, TimeUnit.SECONDS)
                .build();
    }

    @Bean
    public Cache<String, Object> getGuavaCache() {
        return guavaCache;
    }
}

@Override
public TbSeckillGoods queryGoodsDetailsByCache(Integer id) {
    String cacheKey = SECKILL_GOODS_CACHE_PREFIX + id;
    // 1、先从JVM堆内存中读取数据，使用 guava 缓存
    TbSeckillGoods seckillGoods = (TbSeckillGoods) guavaCacche.getIfPresent(cacheKey);
    if (seckillGoods != null) {
        return seckillGoods;
    }
    // 2、如果JVM堆内存中不存在，则从分布式缓存(redis)中查询
    seckillGoods = redisService.getObjValue(cacheKey);
    if (seckillGoods != null) {
        // 添加进guava缓存
        guavaCacche.put(cacheKey, seckillGoods);
        return seckillGoods;
    }
    // 3、如果分布式缓存(redis)中还没有，则从数据库查询
    seckillGoods = seckillGoodsMapper.selectByPrimaryKey(id);
    if (seckillGoods != null && seckillGoods.getStatus() == 1) {
        // 添加进分布式缓存(redis)中
        redisService.setObjValue(cacheKey, seckillGoods, 30, TimeUnit.MINUTES);
    }
    return seckillGoods;
}

@Api(tags = "秒杀下单模块")
@RestController
@RequestMapping("seckill")
public class SeckillOrderController {
    @Resource
    private UserService userService;
    @Resource
    private SeckillOrderService seckillOrderService;

    @ApiOperation("普通下单操作")
    @GetMapping("/order/{id}/{token}")
    public RestResponse<Boolean> generalKilled(@PathVariable Long id,
                                               @PathVariable String token) {
        BaseUser baseUser = userService.queryUserByToken(token);
        if (baseUser == null) {
            return RestResponse.error(RestResponseCode.TOKEN_OVERTIME);
        }
        return RestResponse.success(seckillOrderService.generalKilled(id, baseUser.getGuid()));
    }
}
@Slf4j
@Service
public class SeckillOrderServiceImpl implements SeckillOrderService {
    @Resource
    private TbSeckillGoodsMapper seckillGoodsMapper;
    @Resource
    private TbSeckillOrderMapper seckillOrderMapper;

    @Override
    public boolean generalKilled(Long id, String userId) {
        // 1、从数据库查询商品数据，并进行校验
        TbSeckillGoods seckillGoods = seckillGoodsMapper.selectByPrimaryKey(id);
        validateSeckillGoods(seckillGoods);
        // 2、扣减库存
        seckillGoods.setStockCount(seckillGoods.getStockCount() - 1);
        // 3、更新库存
        seckillGoodsMapper.updateByPrimaryKeySelective(seckillGoods);
        // 4、下单
        TbSeckillOrder seckillOrder = new TbSeckillOrder()
                .setSeckillId(id)
                .setUserId(userId)
                .setCreatedTime(System.currentTimeMillis())
                .setStatus(0)
                .setMoney(seckillGoods.getCostPrice());

        return seckillOrderMapper.insertSelective(seckillOrder) >= 1;
    }

    private void validateSeckillGoods(TbSeckillGoods seckillGoods) {
        if (seckillGoods == null) {
            throw new SeckillOrderException(RestResponseCode.SEC_GOODS_NOT_EXSISTS);
        }
        if (seckillGoods.getStartTime() > System.currentTimeMillis()) {
            throw new SeckillOrderException(RestResponseCode.SEC_ACTIVE_NOT_START);
        }
        if (seckillGoods.getEndTime() <= System.currentTimeMillis()) {
            throw new SeckillOrderException(RestResponseCode.SEC_ACTIVE_END);
        }
        if (seckillGoods.getStatus() != 1) {
            throw new SeckillOrderException(RestResponseCode.SEC_NOT_UP);
        }
        if (seckillGoods.getStockCount() <= 0) {
            throw new SeckillOrderException(RestResponseCode.SEC_GOODS_END);
        }
    }
}

/**
 * 互斥锁，参数默认false:不公平锁
 */
private Lock lock = new ReentrantLock(true);
@Transactional
@Override
public boolean generalKilled(Long id, String userId) {
    boolean result = false;
    lock.lock();
    try {
        // ...
    } finally {
        lock.unlock();
    }
    return result;
}

@Override
public boolean generalKilledByLock(Long id, String userId) {
    boolean result = false;
    lock.lock();
    try {
        result = generalKilled(id, userId);
    } finally {
        lock.unlock();
    }
    return result;
}
@Transactional
@Override
public boolean generalKilled(Long id, String userId) {
	// ...
}

/**
 * 自定义注解，实现aop锁
 */
@Target({ElementType.PARAMETER, ElementType.METHOD})
@Retention(RetentionPolicy.RUNTIME)
@Documented
public @interface ServiceLock {
    String description() default "";
}

@Slf4j
@Component
@Scope
@Aspect
@Order(1)
public class LockAspect {
    /**
     * 定义锁对象
     */
    private static Lock LOCK = new ReentrantLock(true);
    /**
     * Service切入点
     */
    @Pointcut("@annotation(com.veli.vshop.seckill.aop.lock.ServiceLock)")
    public void lockAspect() {

    }

    @Around("lockAspect()")
    public Object around(ProceedingJoinPoint joinPoint) {
        // 初始化一个对象
        Object obj = null;
        // 加锁
        LOCK.lock();
        try {
            // 执行业务
            obj = joinPoint.proceed();
        } catch (Throwable cause) {
            log.error(cause.getMessage(), cause);
        } finally {
            // 业务执行结束后，释放锁
            LOCK.unlock();
        }
        return obj;
    }
}

@Update(value = "update tb_seckill_goods set stock_count=stock_count-1, version=version+1 where id = #{id} and version = #{version}")
int updateByPrimaryKeyWithVersion(@Param("id") Long id, @Param("version") Integer version);

@Target({ElementType.PARAMETER, ElementType.METHOD})
@Retention(RetentionPolicy.RUNTIME)
@Documented
public @interface ServiceRedisLock {
    String description() default "";
}

@Slf4j
@Component
@Scope
@Aspect
@Order(1)
public class LockRedisAspect {
    private static final String SECKILL_GOODS_PREFIX = "seckill_goods_lock_";
    /**
     * 定义锁对象
     */
    @Resource
    private HttpServletRequest request;

    /**
     * Service切入点
     */
    @Pointcut("@annotation(com.veli.vshop.seckill.aop.lock.ServiceRedisLock)")
    public void lockAspect() {

    }

    @Around("lockAspect()")
    public Object around(ProceedingJoinPoint joinPoint) {
        // 获取id
        String uri = request.getRequestURI();
        String id = uri.substring(uri.lastIndexOf("/") - 1, uri.lastIndexOf("/"));
        // 初始化一个对象
        Object obj = null;
        // 加锁: 先获取一把锁
        String lockKey = SECKILL_GOODS_PREFIX + id;
        boolean result = RedissonLockUtils.tryLock(lockKey, 3, 10, TimeUnit.SECONDS);

        try {
            if (result) {
                // 执行业务
                obj = joinPoint.proceed();
            }
        } catch (Throwable cause) {
            if (cause instanceof CustomException) {
                throw (CustomException) cause;
            } else {
                log.error(cause.getMessage());
            }
        } finally {
            // 业务执行结束后，释放锁
            if (result) {
                RedissonLockUtils.unlock(lockKey);
            }
        }
        return obj;
    }
}

@Transactional(rollbackFor = Exception.class)
@Override
public boolean redisCacheKilled(Long id, String userId) {
    // 优化一：从缓存中获取秒杀商品数据
    TbSeckillGoods seckillGoods = redisService.getObjValue(SEC_KILL_GOODS_CACHE_PREFIX + id);
    validateSeckillGoods(seckillGoods);
    // 优化二：利用Redis的原子性操作扣减库存，不需要上锁
    boolean result = reduceStock(id);
    if (!result) {
        throw new SeckillOrderException(RestResponseCode.SEC_GOODS_STOCK_FAIL, "扣减库存失败");
    }
    // 下单
    TbSeckillOrder seckillOrder = new TbSeckillOrder()
            .setSeckillId(id)
            .setUserId(userId)
            .setCreatedTime(System.currentTimeMillis())
            .setStatus(0)
            .setMoney(seckillGoods.getCostPrice());
    // 3、异步实现（blockingQueue,disruptor,rocketMQ队列实现异步）
    // 队列实现异步下单操作
    return seckillOrderMapper.insertSelective(seckillOrder) >= 1;
}

private boolean reduceStock(Long id) {
    Long result = redisService.incrInt(SEC_KILL_GOODS_STOCK_CACHE_PREFIX + id, -1);
    if (result > 0) {
        // TODO 发送消息
        //
        return true;
    } else if (result == 0) {
        //
        // 记录标识，表示此商品已经售卖结束
        redisService.setIntValue(SEC_KILL_GOODS_STOCK_END_CACHE_PREFIX + id, 1);
        return true;
    } else {
        // 扣减库存失败
        redisService.incrInt(SEC_KILL_GOODS_STOCK_CACHE_PREFIX + id, 1);
        return false;
    }
}

public class SeckillQueue {
    /**
     * 队列大小
     */
    private static final int MAX_SIZE = 100;
    /**
     * 定义一个队列：用于多线程间下单的队列
     */
    private static final BlockingQueue<TbSeckillOrder> QUEUE = new LinkedBlockingQueue<>(MAX_SIZE);

    private SeckillQueue() {
    }

    /**
     * 静态内部类实现单例
     */
    private static class Holder {
        private static SeckillQueue INSTANCE = new SeckillQueue();
    }

    public static SeckillQueue getInstance() {
        return Holder.INSTANCE;
    }

    /**
     * 生产入队
     * add(e) 队列未满时，返回true；队列满则抛出IllegalStateException(“Queue full”)异常——AbstractQueue
     * put(e) 队列未满时，直接插入没有返回值；队列满时会阻塞等待，一直等到队列未满时再插入。
     * offer(e) 队列未满时，返回true；队列满时返回false。非阻塞立即返回。
     * offer(e, time, unit) 设定等待的时间，如果在指定时间内还不能往队列中插入数据则返回false，插入成功返回true。
     */
    public boolean produce(TbSeckillOrder order) {
        return QUEUE.offer(order);
    }
    /**
     * 消费出队
     * poll() 获取并移除队首元素，在指定的时间内去轮询队列看有没有首元素有则返回，否者超时后返回null
     * take() 与带超时时间的poll类似不同在于take时候如果当前队列空了它会一直等待其他线程调用notEmpty.signal()才会被唤醒
     */
    public TbSeckillOrder consume() throws InterruptedException {
        return QUEUE.take();
    }
    /**
     * 获取队列大小
     */
    public static int getSize() {
        return QUEUE.size();
    }
}

@Slf4j
@Component
public class TaskRunner implements ApplicationRunner {
    private static final ExecutorService EXECUTOR = Executors.newSingleThreadExecutor();
    @Resource
    private SeckillOrderService seckillOrderService;

    @Override
    public void run(ApplicationArguments args) throws Exception {
        // 提交一个任务，一直监听bockingQueue队列
        EXECUTOR.submit(() -> {
            log.info("==>> TaskRunner started");
            while (true) {
                try {
                    TbSeckillOrder order = SeckillQueue.getInstance().consume();
                    if (order != null) {
                    	// 从队列中获取订单，执行下单操作
                        seckillOrderService.generalKilled(order.getSeckillId(), order.getUserId());
                    }
                } catch (Exception e) {
                    log.error(e.getMessage(), e);
                }
            }
        });
    }
}

private boolean reduceStock(Long id) {
    Long result = redisService.incrInt(SEC_KILL_GOODS_STOCK_CACHE_PREFIX + id, -1);
    if (result >= 0) {
        // 扣减库存成功发送消息
        producer.sendSyncStockMsg(id);
        if (result == 0) { // 记录标识，表示此商品已经售卖结束
            redisService.setIntValue(SEC_KILL_GOODS_STOCK_END_CACHE_PREFIX + id, 1);
        }
        return true;
    } else {
        // 扣减库存失败
        redisService.incrInt(SEC_KILL_GOODS_STOCK_CACHE_PREFIX + id, 1);
        return false;
    }
}

consumer.registerMessageListener((MessageListenerConcurrently) (messages, context) -> {
    try {
        for (MessageExt messageExt : messages) {
            String message = new String(messageExt.getBody(), RemotingHelper.DEFAULT_CHARSET);
            // 同步数据库的库存
            seckillGoodsMapper.updateByPrimaryKeyWithLock(Long.valueOf(message));
            log.info("[Consumer] msgID: {}, msgBody: {}", messageExt.getMsgId(), message);
        }
    } catch (Exception e) {
        // 如果出现异常，必须告知消息进行重试
        return ConsumeConcurrentlyStatus.RECONSUME_LATER;
    }
    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
});

@Slf4j
@Component
public class TransactMQProducer {
    private TransactionMQProducer producer = new TransactionMQProducer();
    @Resource
    private RocketmqConfig rocketmqConfig;
    @Resource
    private SeckillOrderService seckillOrderService;
    @Resource
    private TbSeckillGoodsMapper seckillGoodsMapper;

    @PostConstruct
    public void init() {
        try {
            producer.setProducerGroup(rocketmqConfig.getGroupName());
            producer.setNamesrvAddr(rocketmqConfig.getNamesrvAddr());
            producer.setRetryTimesWhenSendFailed(rocketmqConfig.getRetryTimes());
            producer.setVipChannelEnabled(false);
            producer.start();
            addListener();
            log.info("[Producer] Started ...");
        } catch (Exception e) {
            log.error(e.getMessage(), e);
        }
    }

    public void addListener() {
        // 注入一个监听器
        producer.setTransactionListener(new TransactionListener() {
            /**
             * 执行本地业务的方法
             */
            @Override
            public LocalTransactionState executeLocalTransaction(Message message, Object o) {
                Long seckillId = 0L;
                try {
                    // 获取消息内容
                    String msg = new String(message.getBody(), RemotingHelper.DEFAULT_CHARSET);
                    SeckillDto seckillDto = JsonUtils.toObj(msg, SeckillDto.class);
                    if (seckillDto != null) {
                        seckillId = seckillDto.getSeckillId();
                        seckillOrderService.redisCacheKilled(seckillDto.getSeckillId(), seckillDto.getUserId());
                    }
                } catch (SeckillOrderException e) {
                    // 业务处理中出现一个预知的异常，设置事务回滚状态
                    TbSeckillGoods updateObj = new TbSeckillGoods()
                            .setId(Integer.parseInt(seckillId.toString()))
                            .setTransactionStatus(-1);
                    seckillGoodsMapper.updateByPrimaryKeySelective(updateObj);
                    return LocalTransactionState.ROLLBACK_MESSAGE;
                } catch (Exception e) {
                    log.error(e.getMessage(), e);
                    return LocalTransactionState.UNKNOW;
                }
                // 业务执行成功，确定事务提交状态
                return LocalTransactionState.COMMIT_MESSAGE;
            }

            /**
             * 事务状态回查方法
             */
            @Override
            public LocalTransactionState checkLocalTransaction(MessageExt messageExt) {
                try {
                    // 获取消息内容
                    String msg = new String(messageExt.getBody(), RemotingHelper.DEFAULT_CHARSET);
                    SeckillDto seckillDto = JsonUtils.toObj(msg, SeckillDto.class);
                    // 查询事务状态
                    TbSeckillGoods seckillGoods = seckillGoodsMapper.selectByPrimaryKey(seckillDto.getSeckillId());
                    // 根据事务状态，判定事务 commit, rollback, unkown
                    switch (seckillGoods.getTransactionStatus()) {
                        case -1:
                            return LocalTransactionState.ROLLBACK_MESSAGE;
                        case 1:
                            return LocalTransactionState.COMMIT_MESSAGE;
                        default:
                            return LocalTransactionState.UNKNOW;
                    }
                } catch (Exception e) {
                    log.error(e.getMessage(), e);
                }
                return LocalTransactionState.COMMIT_MESSAGE;
            }
        });
    }

    /**
     * 发送消息，使用事务型消息把所有的操作原子化
     */
    public boolean sendTransactionMsg(Long seckillId, String userId) {
        try {
            SeckillDto seckillDto = new SeckillDto(seckillId, userId);
            byte[] bytes = JsonUtils.toStr(seckillDto).getBytes(RemotingHelper.DEFAULT_CHARSET);
            Message message = new Message(rocketmqConfig.getTopic(), bytes);
            producer.sendMessageInTransaction(message, null);
        } catch (Exception e) {
            log.error(e.getMessage(), e);
            return false;
        }
        return true;
    }

    @PreDestroy
    public void destroy() {
        if (producer != null) {
            producer.shutdown();
        }
    }
}

@Slf4j
@Component
public class RocketmqConsumer {
    @Resource
    private TbSeckillGoodsMapper seckillGoodsMapper;
    @Resource
    private RocketmqConfig rocketmqConfig;

    @Bean
    public DefaultMQPushConsumer seckillMqConsumer() {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(rocketmqConfig.getGroupName());
        consumer.setNamesrvAddr(rocketmqConfig.getNamesrvAddr());
        try {
            // 广播模式消费
            consumer.subscribe(rocketmqConfig.getTopic(), "*");
            // 如果是第一次启动，从队列头部开始消费；如果不是第一次启动，从上次消费的位置继续消费
            consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
            consumer.setVipChannelEnabled(false);
            consumer.registerMessageListener((MessageListenerConcurrently) (messages, context) -> {
                try {
                    for (MessageExt messageExt : messages) {
                        String message = new String(messageExt.getBody(), RemotingHelper.DEFAULT_CHARSET);
                        SeckillDto seckillDto = JsonUtils.toObj(message, SeckillDto.class);
                        // 同步数据库的库存
                        seckillGoodsMapper.updateByPrimaryKeyWithLock(seckillDto.getSeckillId());
                        log.info("[Consumer] msgID: {}, msgBody: {}", messageExt.getMsgId(), JsonUtils.toStr(seckillDto));
                    }
                } catch (Exception e) {
                    // 如果出现异常，必须告知消息进行重试
                    return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                }
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            });
            consumer.start();
            log.info("[Consumer] Started ...");
        } catch (Exception e) {
            log.error(e.getMessage(), e);
        }
        return consumer;
    }
}

@Transactional(rollbackFor = Exception.class)
@Override
public boolean redisCacheKilled(Long id, String userId) {
    // 优化一：从缓存中获取秒杀商品数据
    TbSeckillGoods seckillGoods = redisService.getObjValue(SEC_KILL_GOODS_CACHE_PREFIX + id);
    validateSeckillGoods(seckillGoods);
    // 优化二：利用Redis的原子性操作扣减库存，不需要上锁
    boolean result = reduceStock(id);
    if (!result) {
        throw new SeckillOrderException(RestResponseCode.SEC_GOODS_STOCK_FAIL, "扣减库存失败");
    }
    // 下单
    TbSeckillOrder seckillOrder = new TbSeckillOrder()
            .setSeckillId(id)
            .setUserId(userId)
            .setCreatedTime(System.currentTimeMillis())
            .setStatus(0)
            .setMoney(seckillGoods.getCostPrice());
    // 优化三：异步实现（blockingQueue,disruptor,rocketMQ队列实现异步）
    // 队列实现异步下单操作
    boolean produceRes = SeckillQueue.getInstance().produce(seckillOrder);
    if (!produceRes) {
        throw new SeckillOrderException(RestResponseCode.SEC_GOODS_STOCK_FAIL);
    }
    // 设置事务状态
    seckillGoods.setTransactionStatus(1).setStockCount(null);
    // 更新事务状态
    seckillGoodsMapper.updateByPrimaryKeySelective(seckillGoods);
    return true;
}

// SeckillOrderController
@ApiOperation("⑥ 使用MQ异步下单，实现库存控制和缓存优化")
@GetMapping("/order/mq/{id}/{token}")
public RestResponse<Boolean> mqKilled(@PathVariable Long id,
                                      @PathVariable String token) {
    BaseUser baseUser = userService.queryUserByToken(token);
    if (baseUser == null) {
        return RestResponse.error(RestResponseCode.TOKEN_OVERTIME);
    }
    return RestResponse.success(seckillOrderService.mqKilled(id, baseUser.getGuid()));
}
// SeckillOrderService    
@Override
public boolean mqKilled(Long id, String userId) {
    Future<Object> future = TaskUtils.submit(() -> {
        boolean result = transactMQProducer.sendTransactionMsg(id, userId);
        if (!result) {
            throw new SeckillOrderException();
        }
        return null;
    });
    try {
        future.get();
    } catch (Exception e) {
        throw new SeckillOrderException(RestResponseCode.SEC_GOODS_STOCK_FAIL, "消息发送失败");
    }
    return true;
}