并发新构件之PriorityBlockingQueue:优先阻塞队列

Posted 2020-11-16 houj

PriorityBlockingQueue：优先阻塞队列；是带有优先级的阻塞队列，一个无界阻塞队列，它使用与类 PriorityQueue 相同的顺序规则，并且提供了阻塞获取操作。虽然此队列逻辑上是无界的，但是资源被耗尽时试图执行 add 操作也将失败（导致 OutOfMemoryError）。此类不允许使用 null 元素。依赖自然顺序的优先级队列也不允许插入不可比较的对象（这样做会导致抛出 ClassCastException）。

package com.houjun.current.newClassBank;

import java.util.ArrayList;
import java.util.List;
import java.util.Queue;
import java.util.Random;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.PriorityBlockingQueue;
import java.util.concurrent.TimeUnit;

/**
 * @Author: HouJun
 * @Date: 2019/10/18 8:31
 * @Description: 测试优先阻塞队列
 * @version: 1.0
 */
public class PriorityBlockingQueueDemo {
    public static void main(String[] args) throws InterruptedException {
        ExecutorService exec = Executors.newCachedThreadPool();
        PriorityBlockingQueue<Runnable> priorityBlockingQueue = new PriorityBlockingQueue<>();
        exec.execute(new PrioritizedTaskProducer(priorityBlockingQueue, exec));
//        TimeUnit.SECONDS.sleep(4);
        exec.execute(new PrioritizedTaskConsumer(priorityBlockingQueue));

    }
}

//可加入优先队列的对象，
class PrioritizedTask implements Runnable, Comparable<PrioritizedTask> {
    Random random = new Random(47);
    private static int counter = 0;
    private final int id = counter++;
    private final int priority;
    protected static List<PrioritizedTask> sequence = new ArrayList<>();

    public PrioritizedTask(int priority) {
        this.priority = priority;
        sequence.add(this);
    }


    @Override
    public void run() {
        try {
            TimeUnit.MILLISECONDS.sleep(random.nextInt(250));
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        System.out.println(this);
    }

    @Override
    public String toString() {
        return String.format("[%1$-3d]", priority) + " Task " + id;
    }

    public String summary() {
        return "(" + id + " : " + priority + ")";
    }

    @Override//这样就是升序
    public int compareTo(PrioritizedTask o) {
        if (priority > o.priority)//这样就是升序
            return 1;
        if (priority < o.priority)//这样就是升序
            return -1;
        return 0;
    }
}

//生产者，向优先队列中添加对象
class PrioritizedTaskProducer implements Runnable {
    Random random = new Random(47);
    private Queue<Runnable> queue;
    private ExecutorService executorService;

    public PrioritizedTaskProducer(Queue<Runnable> queue, ExecutorService executorService) {
        this.queue = queue;
        this.executorService = executorService;
    }

    @Override
    public void run() {
        for (int i = 0; i < 20; i++) {
            queue.add(new PrioritizedTask(random.nextInt(10)));
            Thread.yield();
        }
        for (int i = 0; i < 10; i++) {
            try {
                TimeUnit.MILLISECONDS.sleep(250);
                queue.add(new PrioritizedTask(20));
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
        for (int i = 0; i < 10; i++) {
            queue.add(new PrioritizedTask(i));
        }
        System.out.println("Finished PrioritizedTaskProducer");
    }
}

//消费者，取出优先队列的类
class PrioritizedTaskConsumer implements Runnable {
    private PriorityBlockingQueue<Runnable> queue;

    public PrioritizedTaskConsumer(PriorityBlockingQueue<Runnable> queue) {
        this.queue = queue;
        System.out.println("队列中的元素个数" + queue.size());
    }

    @Override
    public void run() {
        while (!Thread.interrupted()) {
            try {
                queue.take().run();//拿出优先级队列中的任务，并运行
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
        System.out.println("Finished PrioritizedTaskConsumer");
    }
}