jihaichannel / demo Goto Github PK

//  真实场景中可能会在热促前进行库存锁定
private Integer skuCount = 10;

private BlockingQueue<RequestPromise> queue = new LinkedBlockingQueue<>(15);

public static void main(String[] args) throws InterruptedException {
    ExecutorService exec = Executors.newCachedThreadPool();
    KillExplain killExplain = new KillExplain();
    killExplain.mergeJob();
    Thread.sleep(2000);

    List<Future<Result>> futureList = new ArrayList<>();
    CountDownLatch latch = new CountDownLatch(15);
    for (int i = 0; i < 15; i++){
        final Long orderId = i + 10L;
        final Long userId = Long.valueOf(i);
        Future<Result> future = exec.submit(() -> {
            latch.countDown();
            return killExplain.buy(new UserRequestBuy(orderId, userId, 1));
        });
        futureList.add(future);
    }

    //  获取执行数据
    futureList.forEach(future -> {
        try {
            Result result = future.get(300, TimeUnit.MILLISECONDS);
            System.out.println("客户端响应：" + result);
        }catch (Exception e){
            e.printStackTrace();
        }
    });
}


/**
 * 库存扣减
 * @param requestBuy
 * @return
 */
private Result buy(UserRequestBuy requestBuy) throws InterruptedException {

    RequestPromise promise = new RequestPromise(requestBuy);
    //  入队
    boolean enterQueue = queue.offer(promise, 100, TimeUnit.MILLISECONDS    );

    //  入队失败，返回
    if (! enterQueue){
        return new Result(false, "Server is busy");
    }
    //  入队成功
    synchronized (promise){
        try {
            //  等待两百毫秒，让mergeJob进行处理
            promise.wait(200);
            if (promise.getResult() == null){
                return new Result(false, "Time out");
            }
        }catch (InterruptedException e){
            return new Result(false, "被中断");
        }
    }

    return promise.getResult();
}

public void mergeJob(){
    new Thread(() -> {
        List<RequestPromise> list = new ArrayList<>();
        while (true){
            //  阻塞队列为空，说明没有buy请求,等待两百毫秒后轮询
            if (queue.isEmpty()){
                try {
                    Thread.sleep(200);
                    continue;
                }catch (InterruptedException e){
                    e.printStackTrace();
                }
            }

            //  双重检查，如果阻塞队列不为空则取出元素
            while(queue.peek() != null){
                list.add(queue.poll());
            }
            System.out.println(Thread.currentThread().getName() + "合并扣减库存" + list);

            /*  =====
                两种情况：
             a.队列中所有请求购买的总数小于库存总数，则批量进行扣减并返回
             b.队列中所有请求购买的总数大于库存总数，则需要遍历列表，进行逐一扣减（海哥视频中有说到要优先考虑大客户的需求，所以可以加上排序，先扣减数量多的，再往下扣减）
             */

            /* =======  a   =======*/
            //  获取当前List中请求buy的数量
            int sum = list.stream().mapToInt(e -> e.getRequestBuy().getCount()).sum();
            //  当buy的数量小于sku总数，先返回成功购买的信息，批量插入后再进行扣减
            if (sum <= skuCount){
                skuCount-= sum;
                list.forEach(requestPromise -> {
                    requestPromise.setResult(new Result(true, "success"));
                    synchronized (requestPromise){
                        //  唤醒执行成功后的当前线程
                        requestPromise.notify();
                    }
                });
                continue;
            }
            /* =======  b   =======*/
            for (RequestPromise requestPromise : list){
                //  获取当前buy请求购买的数量
                int count = requestPromise.getRequestBuy().getCount();
                //  判断是否有足够的库存
                if (count <= skuCount){
                    //  库存扣减
                    skuCount -= count;
                    requestPromise.setResult(new Result(true, "Success"));
                    synchronized (requestPromise){
                        //  唤醒执行成功后的当前线程，让线程wait时间不必等于200毫秒
                        requestPromise.notify();
                    }
                }else {
                    requestPromise.setResult(new Result(false, "库存不足"));
                }
            }
            list.clear();
        }
    }, "MergeThread  ").start();
}