java ThreadPoolExecutor 线程池构造方法参数：队列（ArrayBlockingQueue、LinkedBlockingQueue）

作者：马育民 • • 阅读：10054

关于队列概念

详见：JAVA 多线程中的队列分类：阻塞队列，有界队列，无界队列

常用队列

  • ArrayBlockingQueue： 有界队列，基于数组结构，按照队列FIFO原则对元素排序；

  • LinkedBlockingQueue：无界、无界队列，基于链表结构，按照队列FIFO原则对元素排序，效率高，推荐
    Executors.newFixedThreadPool() 使用了这个队列

  • SynchronousQueue：同步队列, 该队列不存储元素，每个插入操作必须等待另一个线程调用移除操作，否则插入操作会一直被阻塞，
    Executors.newCachedThreadPool() 使用了这个队列；

  • PriorityBlockingQueue：支持优先级排序的无界阻塞队列。

  • DelayQueue： 一个使用优先级队列实现的无界阻塞队列

  • LinkedTransferQueue： 一个由链表结构组成的无界阻塞队列

ArrayBlockingQueue

基于数组实现的有界阻塞队列，ArrayBlockingQueue 内部维护这一个定长数组，阻塞队列的大小在初始化时就已经确定了，其后无法更改。

ArrayBlockingQueue 支持公平性和非公平性，默认采用非公平模式，可以通过构造函数设置为公平访问策略。

LinkedBlockingDeque

LinkedBlockingDeque是一个有链表组成的双向阻塞队列，与前面的阻塞队列相比它支持从两端插入和移出元素。以first结尾的表示从对头操作，以last结尾的表示从对尾操作。

在初始化LinkedBlockingDeque时可以初始化队列的容量，用来防止其再扩容时过渡膨胀。另外双向阻塞队列可以运用在“工作窃取”模式中。

ArrayBlockingQueue VS LinkedBlockingDeque

ArrayBlockingQueue

数据结构：数组，存储空间预先分配，无需动态申请空间，使用过程中内存开销较小；

LinkedBlockingQueue：

数据结构：单项链表，存储空间动态申请，会增加JVM垃圾回收负担；
两把锁，并发性能较好；
可设置为无界，吞吐量比较大，但是不稳定，入队速度太快有可能导致内存溢出。

PriorityBlockingQueue

PriorityBlockingQueue 是支持优先级的无界队列。默认情况下采用自然顺序排序，当然也可以通过自定义Comparator来指定元素的排序顺序，不能保证同优先级元素的顺序。

PriorityBlockingQueue 内部采用二叉堆的实现方式。

SynchronousQueue

SynchronousQueue 是一个神奇的队列，他是一个不存储元素的阻塞队列，也就是说他的每一个put操作都需要等待一个take操作，否则就不能继续添加元素了，有点儿像Exchanger，类似于生产者和消费者进行交换。

队列本身不存储任何元素，所以非常适用于传递性场景，两者直接进行对接。其吞吐量会高于ArrayBlockingQueue和LinkedBlockingQueue。

SynchronousQueue支持公平和非公平的访问策略，在默认情况下采用非公平性，也可以通过构造函数来设置为公平性。

SynchronousQueue的实现核心为Transferer接口，该接口有TransferQueue和TransferStack两个实现类，分别对应着公平策略和非公平策略。接口Transferer有一个tranfer()方法，该方法定义了转移数据，如果e != null，相当于将一个数据交给消费者，如果e == null，则相当于从一个生产者接收一个消费者交出的数据。

例子

本例中，corePoolSize、maximumPoolSize 都是1，workQueue 的大小是2，即：只有1个线程在运行，最多有2个线程放入队列中等待

package ThreadPool;

import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

public class Instance2 {
    public static void main(String[] args) {
        LinkedBlockingQueue linkedBlockingQueue = new LinkedBlockingQueue(2);
        ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(
                1,
                1,
                10,
                TimeUnit.SECONDS,
                linkedBlockingQueue
                );

        print(threadPoolExecutor);
        threadPoolExecutor.execute(new MyRun("1",threadPoolExecutor));

        threadPoolExecutor.execute(new MyRun("2",threadPoolExecutor));

        threadPoolExecutor.execute(new MyRun("3",threadPoolExecutor));

        // 关闭线程池
        threadPoolExecutor.shutdown();
    }
    static void print(ThreadPoolExecutor threadPoolExecutor){

        System.out.println("活动线程数："+threadPoolExecutor.getActiveCount());
        System.out.println("线程池中的线程数："+threadPoolExecutor.getPoolSize());
        System.out.println("队列中的线程数："+threadPoolExecutor.getQueue().size());
        System.out.println("完成的线程数："+threadPoolExecutor.getCompletedTaskCount());
    }
    static class MyRun implements Runnable {
        String name;
        ThreadPoolExecutor threadPoolExecutor;

        public MyRun(String name, ThreadPoolExecutor threadPoolExecutor) {
            this.name = name;
            this.threadPoolExecutor = threadPoolExecutor;
        }

        @Override
        public void run() {
            System.out.println("----"+name+" 线程开始");
            print(threadPoolExecutor);
            try {
                Thread.sleep(3000);
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
            System.out.println("===="+name+" 线程结束");
        }
    }
}

如果提交第4个线程就报错

package ThreadPool;

import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

public class Instance2_超过队列大小报错 {
    public static void main(String[] args) {
        LinkedBlockingQueue linkedBlockingQueue = new LinkedBlockingQueue(2);
        ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(
                1,
                1,
                10,
                TimeUnit.SECONDS,
                linkedBlockingQueue
                );

        print(threadPoolExecutor);
        threadPoolExecutor.execute(new MyRun("1",threadPoolExecutor));

        threadPoolExecutor.execute(new MyRun("2",threadPoolExecutor));

        threadPoolExecutor.execute(new MyRun("3",threadPoolExecutor));

        try {
            threadPoolExecutor.execute(new MyRun("4", threadPoolExecutor));
        }catch (Exception e){
            System.out.println("超过队列大小，抛出异常："+e);
        }
        // 关闭线程池
        threadPoolExecutor.shutdown();
    }
    static void print(ThreadPoolExecutor threadPoolExecutor){

        System.out.println("活动线程数："+threadPoolExecutor.getActiveCount());
        System.out.println("线程池中的线程数："+threadPoolExecutor.getPoolSize());
        System.out.println("队列中的线程数："+threadPoolExecutor.getQueue().size());
        System.out.println("完成的线程数："+threadPoolExecutor.getCompletedTaskCount());
    }
    static class MyRun implements Runnable {
        String name;
        ThreadPoolExecutor threadPoolExecutor;

        public MyRun(String name, ThreadPoolExecutor threadPoolExecutor) {
            this.name = name;
            this.threadPoolExecutor = threadPoolExecutor;
        }

        @Override
        public void run() {
            System.out.println("----"+name+" 线程开始");
            print(threadPoolExecutor);
            try {
                Thread.sleep(3000);
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
            System.out.println("===="+name+" 线程结束");
        }
    }
}

报错如下：

例子2

corePoolSize 是1，maximumPoolSize 是2，队列大小是1

  1. 提交 线程1 时，线程1 开始运行，此时线程池中有1个活动线程

  2. 提交 线程2 时，由于 corePoolSize 是1，此时线程池中有1个活动线程，会将 线程2 放到队列中，此时队列满（队列大小为1）

  3. 提交 线程3 时，由于队列已满，但 maximumPoolSize是2，就继续创建1个线程，并运行

代码：

package ThreadPool;

import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

public class Instance3 {
    public static void main(String[] args) {
        LinkedBlockingQueue linkedBlockingQueue = new LinkedBlockingQueue(1);
        ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(
                1,
                2,
                10,
                TimeUnit.SECONDS,
                linkedBlockingQueue
                );

        print("主线程",threadPoolExecutor);
        threadPoolExecutor.execute(new MyRun("1",threadPoolExecutor));

        threadPoolExecutor.execute(new MyRun("2",threadPoolExecutor));

        threadPoolExecutor.execute(new MyRun("3",threadPoolExecutor));

        // 关闭线程池
        threadPoolExecutor.shutdown();
    }
    static void print(String name,ThreadPoolExecutor threadPoolExecutor){

        System.out.println(name+" 活动线程数："+threadPoolExecutor.getActiveCount());
        System.out.println(name+" 线程池中的线程数："+threadPoolExecutor.getPoolSize());
        System.out.println(name+" 队列中的线程数："+threadPoolExecutor.getQueue().size());
        System.out.println(name+" 完成的线程数："+threadPoolExecutor.getCompletedTaskCount());
    }
    static class MyRun implements Runnable {
        String name;
        ThreadPoolExecutor threadPoolExecutor;

        public MyRun(String name, ThreadPoolExecutor threadPoolExecutor) {
            this.name = name;
            this.threadPoolExecutor = threadPoolExecutor;
        }

        @Override
        public void run() {
            System.out.println("----"+name+" 线程开始");
            print("线程-"+name,threadPoolExecutor);
            try {
                Thread.sleep(3000);
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
            System.out.println("===="+name+" 线程结束");
        }
    }
}

参考：
https://www.jianshu.com/p/b1d383d4021c
参考：
https://blog.csdn.net/u022812849/article/details/114698692

原文出处：/show_1IX4d2dRwbNX.html

吉ICP备15003043号