java高并发案例分析

实体类OrderRequest

package com.hk.springboot.bean;
import java.util.Random;

public class OrderRequest {
    /**
     * 商品id
     */
    private int goodId = new Random().nextInt(100000);
    /**
     * 用户id
     */
    private int userId = new Random().nextInt(100000);
    /**
     * 0:未处理;1:正常;2:异常
     */
    private int status;

    public int getGoodId() {
        return goodId;
    }
    public void setGoodId(int goodId) {
        this.goodId = goodId;
    }
    public int getUserId() {
        return userId;
    }
    public void setUserId(int userId) {
        this.userId = userId;
    }
    public int getStatus() {
        return status;
    }
    public void setStatus(int status) {
        this.status = status;
    }
}

HighConcurrencyController并发控制器

package com.hk.springboot.controller;

import com.hk.springboot.Thread.OrderDealThread;
import com.hk.springboot.bean.OrderRequest;
import com.hk.springboot.common.Constents;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * 高并发demo
 */
@RestController
public class HighConcurrencyController {

    private static final Logger logger = LoggerFactory.getLogger(HighConcurrencyController.class);

    private static AtomicInteger totalOrders;
    private static AtomicInteger totalQueueSize;
    private static ConcurrentLinkedQueue<OrderRequest> orderQueue = new ConcurrentLinkedQueue<OrderRequest>();
    private ExecutorService excutorService = Executors.newCachedThreadPool();
    //public static ReentrantLock queueLock = new ReentrantLock(true);

    static {
        totalOrders = new AtomicInteger(Constents.TOTAL_QUEUE_SIZE);
        totalQueueSize = new AtomicInteger(Constents.TOTAL_QUEUE_SIZE);
    }

    @RequestMapping("/highConcurrency")
    public String highConcurrency(OrderRequest orderRequest) {
        try {
            // queueLock.lock();
            logger.info("highConcurrency==>>start");
            if (totalOrders.get() < 1) {
                //如果库存不足
                logger.info("库存不足==>>end");
                return "error";
            } else {
                //如果总的队列没有满的话，则还可以将新的请求放入进去
                if (orderQueue.size() < totalQueueSize.get()) {
                    logger.info(orderRequest.getGoodId()+" 增加到待处理队列成功:" + orderQueue.size());
                    orderQueue.add(orderRequest);
                } else {
                    //否则队列已满，不能再入队列
                    logger.info("queue full==>>end");
                    return "error";
                }
            }
            //如果线程类的当前执行标志为未执行,即空闲状态,通过线程池启动线程
            if (!OrderDealThread.dealLock.isLocked()) {
                OrderDealThread dealQueue = new OrderDealThread(orderQueue);
                excutorService.execute(dealQueue);
            }
        } catch (Exception e) {
            e.printStackTrace();
            return "not done";
        } finally {
            // queueLock.unlock();
        }
        return "ok";
    }
}

线程处理类OrderDealThread

package com.hk.springboot.Thread;


import com.hk.springboot.bean.OrderRequest;
import com.hk.springboot.common.Constents;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Iterator;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.ReentrantLock;

/**
 * 线程处理类
 */
public class OrderDealThread implements Runnable {

    private static final Logger logger = LoggerFactory.getLogger(OrderDealThread.class);

    private ConcurrentLinkedQueue<OrderRequest> orderQueue;
    private static AtomicInteger totalOrders;
    public static ReentrantLock dealLock = new ReentrantLock(true);

    static {
        totalOrders = new AtomicInteger(Constents.TOTAL_QUEUE_SIZE);
    }

    public OrderDealThread(ConcurrentLinkedQueue<OrderRequest> queque) {
        this.orderQueue = queque;
    }

    @Override
    public void run() {
        while (!orderQueue.isEmpty()) {
            try {
                dealLock.lock();
                Iterator<OrderRequest> it = orderQueue.iterator();
                while (it.hasNext()) {
                    dealQueque(it.next());
                }
            } catch (Exception e) {
                e.printStackTrace();
            } finally {
                dealLock.unlock();
            }
        }
    }

    void dealQueque(OrderRequest orderRequest) {
        if (orderRequest.getStatus() == 0) {
            int status = 2;
            /**
             * 需再次判断是否还有商品，加锁范围内
             */
            if (totalOrders.get() > 0) {
                logger.info("=======do -1=======");
                totalOrders.decrementAndGet();// 减库存
                status =1;
            }
            if (status == 2) {
                logger.info(orderRequest.getUserId() + " deal er:" + Thread.currentThread().getName());
                orderRequest.setStatus(2);
            } else {
                logger.info(orderRequest.getUserId() + " deal ok:" + Thread.currentThread().getName());
                orderRequest.setStatus(1);
            }
        }
    }
}