Java中的阻塞队列BlockingQueue使用详解

/**
 * @Author Dominick Li
 * @CreateTime 2022/3/6 20:03
 * @Description 消息生产者
 **/
public class Product implements Runnable {
    private BlockingQueue<String> bq;
    /**
     * 多少秒生产一条任务
     */
    private int period;
    private Random r = new Random();
    /**
     * 生产者名称
     */
    private String name;
    Product(BlockingQueue<String> bq, int period, String name) {
        this.bq = bq;
        this.period=period;
        this.name=name;
    }
    @Override
    public void run() {
        try {
            while (true){
                Thread.sleep(period);
                String product=String.valueOf(r.nextInt(100));
                //如果队列满了则阻塞
                bq.put(product);
                System.out.println("生产者["+this.name+"]生产"+product+"，当前队列中产品为:"+bq);
            }
        }catch (Exception e){
            e.printStackTrace();
        }
    }
}
/**
 * @Author Dominick Li
 * @CreateTime 2022/3/6 20:11
 * @Description 消费者
 **/
public class Cusumer implements Runnable {
    private BlockingQueue<String> bq;
    /**
     * 多少秒获取一条任务
     */
    private int period;
    /**
     * 消费者名称
     */
    private String name;
    Cusumer(BlockingQueue<String> bq, int period, String name) {
        this.bq = bq;
        this.period=period;
        this.name=name;
    }
    @Override
    public void run() {
        try {
            while (true){
                Thread.sleep(period);
                String value=bq.take();
                System.out.println("消费者["+this.name+"]消费"+value+"，当前队列中产品为:"+bq);
            }
        }catch (Exception e){
            e.printStackTrace();
        }
    }
}
public class Test {
    public static void main(String[] args) {
        BlockingQueue blockingQueue = new ArrayBlockingQueue(5);
        ExecutorService pool = Executors.newCachedThreadPool();
        pool.execute(new Product(blockingQueue, 1000, "生产者"));
        pool.execute(new Cusumer(blockingQueue, 5000, "消费者001"));
        pool.execute(new Cusumer(blockingQueue, 5000, "消费者002"));
        pool.shutdown();
    }
}

public class Test {
    public static void main(String[] args) {
        LinkedBlockingQueue linkedBlockingQueue = new LinkedBlockingQueue(5);
        ExecutorService pool = Executors.newCachedThreadPool();
        pool.execute(new Product(linkedBlockingQueue, 1000, "生产者"));
        pool.execute(new Cusumer(linkedBlockingQueue, 5000, "消费者1"));
        pool.shutdown();
    }
}

public class PriorityProduct implements Comparable<PriorityProduct> {
    /**
     * 任务的优先级
     */
    private int priority;
    private String productName;
    public PriorityProduct(int priority, String productName) {
        this.priority = priority;
        this.productName = productName;
    }
    @Override
    public int compareTo(PriorityProduct o) {
        if (o == null) return -1;
        if (o == this) return 0;
        return o.priority - this.priority;
    }
    @Override
    public String toString(){
        return "{priority="+priority+",name="+this.productName;
    }
}
public class PriorityBlockQueueProduct implements Runnable {
    private PriorityBlockingQueue<PriorityProduct> bq;
    /**
     * 多少秒生产一条任务
     */
    private int period;
    private Random r = new Random();
    public PriorityBlockQueueProduct(PriorityBlockingQueue<PriorityProduct> bq, int period) {
        this.bq = bq;
        this.period = period;
    }
    @Override
    public void run() {
        try {
            while (true) {
                Thread.sleep(period);
                if(bq.size()>10){
                    //限制大小
                    continue;
                }
                  //随机生成优先级5以内的
                PriorityProduct priorityProduct = new PriorityProduct(r.nextInt(5), "test");
                //如果队列满了则阻塞
                bq.put(priorityProduct);
                //System.out.println("生产者商品[" +priorityProduct + "]，当前队列中产品为:" + bq);
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}
public class PriorityBlockQueueCusumer implements Runnable {
    private PriorityBlockingQueue<PriorityProduct> bq;
    /**
     * 多少秒消费一条任务
     */
    private int period;
    public PriorityBlockQueueCusumer(PriorityBlockingQueue<PriorityProduct> bq, int period) {
        this.bq = bq;
        this.period = period;
    }
    @Override
    public void run() {
        try {
            while (true) {
                Thread.sleep(period);
                //如果队列满了则阻塞
                PriorityProduct priorityProduct=bq.take();
                System.out.println("消费产品[" +priorityProduct + "]，当前队列中产品为:" + bq);
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}
public class PriorityTest {
    public static void main(String[] args) {
        PriorityBlockingQueue<PriorityProduct> priorityProducts=new PriorityBlockingQueue<>();
        ExecutorService executorService= Executors.newFixedThreadPool(2);
        executorService.execute(new PriorityBlockQueueProduct(priorityProducts,100));
        executorService.execute(new PriorityBlockQueueCusumer(priorityProducts,1000));
    }
}