您的位置: 首页  >  文章  >  线程池

线程池

分类: 文章 • 2025-01-20 13:55:22

学习笔记：

合理的利用线程池能带来很多好处，如降低资源消耗（通过线程复用，减少线程创建和销毁的消耗），提供响应速度（不需要等待线程创建，可以直接执行任务），易于线程管理（线程池可以统一管理、分配、修改参数等）。

java的线程池主要是通过ThreadPoolExecutor来实现。我们使用的ExecutorService的各种线程池策略都是基于ThreadPoolExecutor实现的。

线程池

线程执行步骤：

第一步：调用ThreadPoolExecutor的execute提交线程，首先检查CotePool大小，如果核心线程数量大小小于CorePoolSize,新创建线程执行任务。

第二步：如果当前CorePool内的线程大于等于corePoolSize，name将线程加入到BlockingQueue中。

第三步：如果不能加入BlockingQueue，在小于MaxPoolSize的情况下创建线程执行任务。

第四步：如果线程数大于等于MaxPoolSize，则执行拒绝策略（RejectedExecutionHandler）。

创建线程池

/**
 * Creates a new {@code ThreadPoolExecutor} with the given initial
 * parameters.
 *
 * @param corePoolSize the number of threads to keep in the pool, even
 *        if they are idle, unless {@code allowCoreThreadTimeOut} is set
 * @param maximumPoolSize the maximum number of threads to allow in the
 *        pool
 * @param keepAliveTime when the number of threads is greater than
 *        the core, this is the maximum time that excess idle threads
 *        will wait for new tasks before terminating.
 * @param unit the time unit for the {@code keepAliveTime} argument
 * @param workQueue the queue to use for holding tasks before they are
 *        executed.  This queue will hold only the {@code Runnable}
 *        tasks submitted by the {@code execute} method.
 * @param threadFactory the factory to use when the executor
 *        creates a new thread
 * @param handler the handler to use when execution is blocked
 *        because the thread bounds and queue capacities are reached
 * @throws IllegalArgumentException if one of the following holds:<br>
 *         {@code corePoolSize < 0}<br>
 *         {@code keepAliveTime < 0}<br>
 *         {@code maximumPoolSize <= 0}<br>
 *         {@code maximumPoolSize < corePoolSize}
 * @throws NullPointerException if {@code workQueue}
 *         or {@code threadFactory} or {@code handler} is null
 */
public ThreadPoolExecutor(int corePoolSize,
                          int maximumPoolSize,
                          long keepAliveTime,
                          TimeUnit unit,
                          BlockingQueue<Runnable> workQueue,
                          ThreadFactory threadFactory,
                          RejectedExecutionHandler handler) {
    if (corePoolSize < 0 ||
        maximumPoolSize <= 0 ||
        maximumPoolSize < corePoolSize ||
        keepAliveTime < 0)
        throw new IllegalArgumentException();
    if (workQueue == null || threadFactory == null || handler == null)
        throw new NullPointerException();
    this.acc = System.getSecurityManager() == null ?
            null :
            AccessController.getContext();
    this.corePoolSize = corePoolSize;
    this.maximumPoolSize = maximumPoolSize;
    this.workQueue = workQueue;
    this.keepAliveTime = unit.toNanos(keepAliveTime);
    this.threadFactory = threadFactory;
    this.handler = handler;
}

参数说明：

    √corePoolSize 核心线程池大小（默认情况下，线程池中的线程数为0，当有任务到来后，创建新线程执行任务，当线程池中的线程数目到达corePoolSize后，就会把到达的任务放到缓存队列当中。）

  √maximumPoolSize 线程池最大容量大小（字面意思，newCachedThreadPool下要注意，它的默认值为Integer.MAX_VALUE，在没有空闲线程可回收利用时，会无限创建线程，直到JVM内存吃满，堆内存溢出 ，需要手动设置）

    √keepAliveTime 线程池空闲时，线程存活时间（默认情况下，只有当线程池中的线程数大于corePoolSize时，某个线程空闲的时间达到keepAliveTime 后，会被销毁，直到线程池中的线程数不超过核心线程池带下。newFixedThreadPool下0秒，newCachedThreadPool下60秒）

    √unit时间单位（7种，天，小时，分钟，秒，毫秒，微妙，纳秒）

    √threadFactory 线程工厂 （defaultThreadFactory和privilegedThreadFactory（默认的是用户线程，优先级为5），可以自定义，利于管理）

 √Blocking 任务队列（阻塞队列，用来存储等待执行的任务，一般三种，ArrayBlockingQueue,LinkedBlockingQueue,SynchronousQueue）

    √RejectedExecutionHandler 线程拒绝策略（四种）

---------------------------------------------------

线程池的层次关系

线程池

   

对于ThreadFactory ，最好自定义线程工厂

好处：可以方便管理，设置更有意义的线程名，自己选择线程类型（用户线程或者守护线程），设置线程优先级，捕获异常等

public class NamedThreadFactory implements ThreadFactory {
    private static final AtomicInteger poolNumber = new AtomicInteger(1);
    private final ThreadGroup group;
    private final AtomicInteger threadNumber = new AtomicInteger(1);
    private final String namePrefix;
    private final boolean daemon;

    public NamedThreadFactory() {
        this("pool-" + poolNumber.getAndIncrement() + "-thread-", false);
    }

    public NamedThreadFactory(String prefix) {
        this(prefix, false);
    }

    public NamedThreadFactory(String namePrefix, boolean daemon) {
        this.namePrefix = StringUtils.isNotEmpty(namePrefix) ? (namePrefix + "-thread-") : ("pool-" + poolNumber.getAndIncrement() + "-thread-");
        this.daemon = daemon;
        SecurityManager security = System.getSecurityManager();
        // 返回线程组，该线程组在调用该线程时实例化正在创建的任何新线程。默认情况下，它返回当前线程的线程组。这应该被特定的安全管理器覆盖，以返回适当的线程组。
        group = Objects.isNull(security) ? Thread.currentThread().getThreadGroup() : security.getThreadGroup();
    }

    @Override
    public Thread newThread(Runnable r) {
        String name = namePrefix + threadNumber.getAndIncrement();
        Thread thread = new Thread(group, r, name, 0);
        thread.setDaemon(daemon);
        return thread;
    }
}

线程池工具类

推荐使用ThreadPoolExecutor,因为Executors创建的线程池由于默认的任务队列是无界的，有发生OOM的可能性

public class ThreadPoolUtil {

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

    /**
     * 线程池容量
     */
    private static int threadPoolCount = Runtime.getRuntime().availableProcessors();
    /**
     * 线程工厂
     */

    private static ThreadFactory threadFactory = new NamedThreadFactory("线程池");
    /**
     * 线程池，例子使用的是Executors，但它创建的线程池是有问题的，有发生OOM的可能性，推荐使用ThreadPoolExecutor
     */
    private static ExecutorService executorService = Executors.newFixedThreadPool(threadPoolCount, threadFactory);

    /**
     * 执行无返回值的方法
     *
     * @param obj        要执行方法所属对象
     * @param methodName 方法签名
     * @param args       方法参数
     */
    public static void execute(Object obj, String methodName, Object... args) {
        executorService.execute(new Task(null, obj, methodName, args));
    }

    public static void execute(CountDownLatch countDownLatch, Object obj, String methodName, Object... args) {
        executorService.execute(new Task(countDownLatch, obj, methodName, args));
    }

    /**
     * 执行有返回值的方法
     *
     * @param obj        要执行方法所属对象
     * @param methodName 方法签名
     * @param args       方法参数
     * @param <T>
     * @return
     */
    public static <T> T submit(Object obj, String methodName, Object... args) {
        Future<T> future = executorService.submit(new Call<T>(null, obj, methodName, args));
        try {
            return future.get();
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (ExecutionException e) {
            e.printStackTrace();
        }
        return null;
    }

    public static <T> T submit(CountDownLatch countDownLatch, Object obj, String methodName, Object... args) {
        Future<T> future = executorService.submit(new Call<T>(countDownLatch, obj, methodName, args));
        try {
            return future.get();
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (ExecutionException e) {
            e.printStackTrace();
        }
        return null;
    }


    /**
     * 有返回值
     *
     * @param <T>
     */
    private static class Call<T> implements Callable {
        private CountDownLatch countDownLatch;
        private Object obj;
        private String methodName;
        private Object[] args;

        public Call(CountDownLatch countDownLatch, Object obj, String methodName, Object... args) {
            this.obj = obj;
            this.countDownLatch = countDownLatch;
            this.methodName = methodName;
            this.args = args;
        }

        @Override
        public T call() {
            System.out.println(Thread.currentThread().getName());
            try {
                Method method = obj.getClass().getDeclaredMethod(methodName, getParamerClass(args));
                if (!method.isAccessible()) {
                    method.setAccessible(true);
                }
                return (T) method.invoke(obj, args);
            } catch (NoSuchMethodException e) {
                logger.error("目标对象{}中没有方法签名{}", obj.getClass().getName(), methodName);
                e.printStackTrace();
            } catch (IllegalAccessException e) {
                logger.error("目标方法{}无法被访问", methodName);
                e.printStackTrace();
            } catch (InvocationTargetException e) {
                logger.error("目标对象{}的{}方法反射调用异常", obj.getClass().getName(), methodName);
                e.printStackTrace();
            } finally {
                if (Objects.nonNull(countDownLatch)) {
                    countDownLatch.countDown();
                }
            }
            return null;
        }
    }

    /**
     * 无返回值
     */
    private static class Task implements Runnable {
        private CountDownLatch countDownLatch;
        private Object obj;
        private String methodName;
        private Object[] args;

        public Task(CountDownLatch countDownLatch, Object obj, String methodName, Object... args) {
            this.obj = obj;
            this.countDownLatch = countDownLatch;
            this.methodName = methodName;
            this.args = args;
        }

        @Override
        public void run() {
            logger.info(Thread.currentThread().getName()+"线程");
            try {
                Method method = obj.getClass().getDeclaredMethod(methodName, getParamerClass(args));
                if (!method.isAccessible()) {
                    method.setAccessible(true);
                }
                method.invoke(obj, args);
            } catch (NoSuchMethodException e) {
                logger.error("目标对象{}中没有方法签名{}", obj.getClass().getName(), methodName);
                e.printStackTrace();
            } catch (IllegalAccessException e) {
                logger.error("目标方法{}无法被访问", methodName);
                e.printStackTrace();
            } catch (InvocationTargetException e) {
                logger.error("目标对象{}的{}方法反射调用异常", obj.getClass().getName(), methodName);
                e.printStackTrace();
            } finally {
                if (Objects.nonNull(countDownLatch)) {
                    countDownLatch.countDown();
                }
            }
        }
    }

    private static Class[] getParamerClass(Object[] args) {
        if (ArrayUtils.isEmpty(args)) {
            return new Class[]{Void.class};
        }
        int length = args.length;
        Class[] classes = new Class[length];
        for (int i = 0; i < length; i++) {
            classes[i] = args[i].getClass();
        }
        return classes;
    }

    static class Test {
        public Integer test(Integer i1, Integer i2) {
            Integer result = i1 + i2;
            System.out.println(result);
            return result;
        }
    }

    public static void main(String[] args) {
        Test test = new Test();
        CountDownLatch countDownLatch = new CountDownLatch(2);
        Integer integer = ThreadPoolUtil.submit(countDownLatch, test, "test", 1, 2);
        Integer integer2 = ThreadPoolUtil.submit(countDownLatch, test, "test", 1, 3);
        System.out.println(integer);
        System.out.println(integer2);
    }
}

工具类里：

execute和submit的区别

1.接收的参数不用，execute接收Runnable类型，而subimt接收Runnable和Callable类型的

2.submit有返回值，Future类型

3.submit方便异常处理（在spring框架里，在事务里使用了线程池，有异常需要回滚的话，这里可以手动抛出一个异常）

接下来整理以下笔记

1.线程的基础笔记

2.几种线程池区别

3.CountDownLatch、CyclicBarrier和Semaphore信号量的区别



相关推荐