java PriorityBlockingQueue的使用

本教程操作环境：windows7系统、java10版，DELL G3电脑。

1.概念

使用平衡二叉树堆，实现的具有优先级的无界阻塞队列。是一个BlockingQueue，所以它是线程安全的。

2.特点

（1）无边界设计，但容量实际是依靠系统资源影响

（2）添加元素，如果超过1，则进入优先级排序

3.应用实例

有任务要执行，可以对任务加一个优先级的权重，这样队列会识别出来，对该任务优先进行出队。

例子中定义了一个将要放入“优先阻塞队列”的任务类，并且定义了一个任务工场类和一个任务执行类，在任务工场类中产生了各种不同优先级的任务，将其添加到队列中，在任务执行类中，任务被一个个取出并执行。

packagecom.niuh.queue.priority;

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

/**
*<p>
*PriorityBlockingQueue使用示例
*</p>
*/
publicclassPriorityBlockingQueueDemo{

publicstaticvoidmain(String[]args)throwsException{
Randomrandom=newRandom(47);
ExecutorServiceexec=Executors.newCachedThreadPool();
PriorityBlockingQueue<Runnable>queue=newPriorityBlockingQueue<>();
exec.execute(newPrioritizedTaskProducer(queue,exec));//这里需要注意，往PriorityBlockingQueue中添加任务和取出任务的
exec.execute(newPrioritizedTaskConsumer(queue));//步骤是同时进行的，因而输出结果并不一定是有序的
}
}

classPrioritizedTaskimplementsRunnable,Comparable<PrioritizedTask>{
privateRandomrandom=newRandom(47);
privatestaticintcounter=0;
privatefinalintid=counter++;
privatefinalintpriority;

protectedstaticList<PrioritizedTask>sequence=newArrayList<>();

publicPrioritizedTask(intpriority){
this.priority=priority;
sequence.add(this);
}

@Override
publicintcompareTo(PrioritizedTasko){
returnpriority<o.priority?1:(priority>o.priority?-1:0);//定义优先级计算方式
}

@Override
publicvoidrun(){
try{
TimeUnit.MILLISECONDS.sleep(random.nextInt(250));
}catch(InterruptedExceptione){
}
System.out.println(this);
}

@Override
publicStringtoString(){
returnString.format("[%1$-3d]",priority)+"Task"+id;
}

publicStringsummary(){
return"("+id+":"+priority+")";
}

publicstaticclassEndSentinelextendsPrioritizedTask{
privateExecutorServiceexec;

publicEndSentinel(ExecutorServiceexec){
super(-1);
this.exec=exec;
}

@Override
publicvoidrun(){
intcount=0;
for(PrioritizedTaskpt:sequence){
System.out.print(pt.summary());
if(++count%5==0){
System.out.println();
}
}
System.out.println();
System.out.println(this+"CallingshutdownNow()");
exec.shutdownNow();
}
}
}

classPrioritizedTaskProducerimplementsRunnable{
privateRandomrandom=newRandom(47);
privateQueue<Runnable>queue;
privateExecutorServiceexec;

publicPrioritizedTaskProducer(Queue<Runnable>queue,ExecutorServiceexec){
this.queue=queue;
this.exec=exec;
}

@Override
publicvoidrun(){
for(inti=0;i<20;i++){
queue.add(newPrioritizedTask(random.nextInt(10)));//往PriorityBlockingQueue中添加随机优先级的任务
Thread.yield();
}
try{
for(inti=0;i<10;i++){
TimeUnit.MILLISECONDS.sleep(250);
queue.add(newPrioritizedTask(10));//往PriorityBlockingQueue中添加优先级为10的任务
}
for(inti=0;i<10;i++){
queue.add(newPrioritizedTask(i));//往PriorityBlockingQueue中添加优先级为1-10的任务
}
queue.add(newPrioritizedTask.EndSentinel(exec));
}catch(InterruptedExceptione){
}
System.out.println("FinishedPrioritizedTaskProducer");
}
}

classPrioritizedTaskConsumerimplementsRunnable{
privatePriorityBlockingQueue<Runnable>queue;

publicPrioritizedTaskConsumer(PriorityBlockingQueue<Runnable>queue){
this.queue=queue;
}

@Override
publicvoidrun(){
try{
while(!Thread.interrupted()){
queue.take().run();//任务的消费者，从PriorityBlockingQueue中取出任务执行
}
}catch(InterruptedExceptione){
}
System.out.println("FinishedPrioritizedTaskConsumer");
}
}

原文来自：https://www.py.cn