线程池------小记

标签：log debug 任务线程 ------ CPU pool 小记

1、线程池的产生背景

1、线程是一种系统资源，每创建一个新的线程都会占用一定的内存。如果是高并发的情况下，短时间生成了很多任务，如果为每个任务都创建一个新的线程，对内存的占用是相当大的，甚至有可能出现内存内存溢出。
2、同时线程也不是创建的越多越好，在cpu核数的限制下，当需要大量的线程进行工作时，cpu必然会让其中获取不到cpu的时间片的线程陷入阻塞，这个就会引起cpu上下文切换的问题，它会先将这部分线程的状态保存下来，轮到它执行的时候，再将其状态恢复过来。上下文切换的越频繁，那么对系统性能就影响越大。

2、线程池

1、线程池是类似于连接池
2、是为了避免频繁的创建和销毁线程
3、系统在启动时就创建大量空闲的线程,当程序将一个Runnable对象或Callable对象传给线程池时,线程池会从中拿出一个线程来执行它们的run()或call()方法,当run()方法或call()方法执行结束时,该线程不会死亡,而是归还给线程池,成为空闲状态,等待下一次被调用.

　自定义线程池

一个线程可以被复用的线程池（消费者），一个阻塞队列（存放任务），主线程（生产任务）。

为什么要使用阻塞队列呢？

阻塞队列可以将因队列长度已满而不能添加到队列的任务先阻塞起来（释放cpu资源），当队列不满时，通过唤醒的方式再将其加入队列。阻塞队列自带阻塞和唤醒功能。

3、线程池的状态

//ThreadPoolExecutor使用int的高3位来表示线程的状态，低29位表示线程数量。
private static final int RUNNING    = -1 << COUNT_BITS;
private static final int SHUTDOWN   =  0 << COUNT_BITS;
private static final int STOP       =  1 << COUNT_BITS;
private static final int TIDYING    =  2 << COUNT_BITS;
private static final int TERMINATED =  3 << COUNT_BITS;

状态名	接收新任务	处理阻塞队列任务	说明	高3位
RUNNING	Y	Y	线程池创建出来，初始状态就是running状态	111
SHUTDOWN	N	Y	当调用shutdown()方法时，正在执行的任务和阻塞队列中的任务都会被执行完，之后线程池的状态才会停止，但是不会去接收新的任务	000
STOP	N	N	当调用shutdownNow()方法时状态会变为stop，它会打断正在执行的任务，并且阻塞队列中的任务也会抛弃掉	001
TIDYING			过渡状态，当所有的任务都执行完了，并且活动的线程数也为0，即将进入终结状态	010
TERMINATED			线程池已经不工作了，处于终结状态	011

它用一个整数来保存线程的状态+线程数量，是为了可以通过一次原子操作便可以改变并保存两种信息。

//原始值 ctlOf(RUNNING, 0)
private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0))
//c为期望值，ctlOf(targetStatus,workerCountOf(c))为更新值，当原始值与期望值相同时，才会进行更新操作
ctl.compareAndSet(c,ctlOf(targetStatus,workerCountOf(c)));

private static int ctlOf(int rs, int wc) { return rs | wc; }

4、构造函数

public ThreadPoolExecutor(int corePoolSize,
                          int maximumPoolSize,
                          long keepAliveTime,
                          TimeUnit unit,
                          BlockingQueue<Runnable> workQueue,
                          ThreadFactory threadFactory,
                          RejectedExecutionHandler handler)

实际上的所有参数为：

corePoolSize：核心线程数（池中最多保留的线程数，一直存在）
maximumPoolSize：最大线程数（池中最多允许存在的线程数）
keepAliveTime：存活时间针对救急线程执行完任务后的存活时间
unit：时间单位针对救急线程
workQueue：阻塞队列（核心线程使用完后会将新的任务加入到阻塞队列中）
threadFactory：线程工厂（创建线程对象，可以起名字，方便与其他线程辨识）
handler：拒绝策略（当最大线程数和阻塞队列都达临界值时，再产生新的任务则触发拒绝策略）

    @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

jdk提供的拒绝策略：

AbortPolicy：让调用这抛出RejectedExecutionException异常，这是默认策略。
CallerRunsPolicy：让调用者运行任务。
DiscardOldestPolicy：放弃队列中最早的任务，本次任务取而代之。
DiscardPolicy：放弃本次任务。

5、线程池类别

5.1 newFixedThreadPool

一个可重用固定线程数的线程池，无界。

public static ExecutorService newFixedThreadPool(int nThreads) {
        return new ThreadPoolExecutor(nThreads, nThreads,
                                      0L, TimeUnit.MILLISECONDS,
                                      new LinkedBlockingQueue<Runnable>());
    }

特点：

1、核心线程数=最大线程数 ，因此意味着无需超时时间，核心线程数永远存活。
2、阻塞队列为LinkedBlockingQueue没有指名capacity，意味着是无界的，可以放任意数量的任务。

适用于：任务量已知，并且相对耗时的任务。

5.2 newCacheThreadPool

带缓冲功能的线程池

public static ExecutorService newCachedThreadPool() {
        return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
                                      60L, TimeUnit.SECONDS,
                                      new SynchronousQueue<Runnable>());
    }

特点：

1、核心线程数为0，最大线程数为Integer.MAX_VALUE，意味着救急线程数为Integer.MAX_VALUE。
2、全部都是救急线程（60s后可以回收）。
3、阻塞队列为SynchronousQueue，没有容量，如果没有线程来取任务是放不进去任务的，同步机制。
4、整个线程池中线程数会根据任务量不断的增长，没有上限，当任务执行完毕，空闲一分钟后释放线程。

适用于：任务数比较密集，但是每个任务执行时间较短。

5.3 newSingleThreadExecutor

单个线程的线程池

  public static ExecutorService newSingleThreadExecutor(ThreadFactory    threadFactory) {
        return new FinalizableDelegatedExecutorService
            (new ThreadPoolExecutor(1, 1,
                                    0L, TimeUnit.MILLISECONDS,
                                    new LinkedBlockingQueue<Runnable>(),
                                    threadFactory));
    }

特点：

1、核心线程数=最大线程数=1。
2、只有一个线程，那么所有任务都必须由一个线程执行，也就意味着业务决定的任务是有顺序性的。
3、可以保证线程池中始终有一个线程。（尽管线程执行时出现异常，队列中的其他任务也会被新的线程重新执行）

与newFixedThreadPool(1)的区别：

newFixedThreadPool返回的是ThreadPoolExecutor线程池对象
newSingleThreadExecutor返回的是FinalizableDelegatedExecutorService对象，该对象包装了线程池对象，但是它只暴露了线程池对象的基本方法（ExecutorService接口中的方法），因此不能调用ThreadPoolExecutor中的特有方法（比如修改核心线程数等方法）。

5.4 任务调度线程池

在JDK1.5时可以使用java.util.Timer来实现定时功能，Timer的优点在于简单易用，但是它的所有任务都是由一个线程执行的，因此所有任务都是串行执行的，同一时间只能有一个任务执行，前一个任务的延迟或者异常还会影响到后面任务的执行。

因此出现任务调度线程池 newScheduledThreadPool

public ScheduledThreadPoolExecutor(int corePoolSize) {
        super(corePoolSize, Integer.MAX_VALUE, 0, NANOSECONDS,
              new DelayedWorkQueue());
    }

方法：
schedule()任务调度不受异常的影响，不受单个任务执行时间长的影响
scheduleAtFixedRate() 如果任务执行的时间过长，那么就不管延时时间了，任务和任务挨着执行
scheduleWithFixedDelay() 本次任务执行完后才进行延时

@Slf4j(topic = "thread.testScheduledThreadPool")
public class TestScheduledThreadPool {
    public static void main(String[] args) {
        //testScheduled();
        testScheduleAtFixedRate();
        //testScheduleWithDelay();

    }


    /**
     * 本次任务执行完后才进行延时
     */
    private static void testScheduleWithDelay(){
        ScheduledExecutorService pool = Executors.newScheduledThreadPool(2);
        log.debug("start....");
        pool.scheduleWithFixedDelay(()->{
            log.debug("task1...");
            try {
                sleep(1000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        },1,2,TimeUnit.SECONDS);
    }

    /**
     *如果任务执行的时间过长，那么就不管延时时间了，任务和任务挨着执行
     */
    private static void testScheduleAtFixedRate() {
        ScheduledExecutorService pool = Executors.newScheduledThreadPool(2);
        log.debug("start....");
        pool.scheduleAtFixedRate(() -> {
            log.debug("task1");
            try {
                sleep(2000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }, 1, 1, TimeUnit.SECONDS);

    }

    /**
     *任务调度不受异常的影响，不受单个任务执行时间长的影响
     */
    private static void testScheduled() {
        ScheduledExecutorService pool = Executors.newScheduledThreadPool(2);
        pool.schedule(() -> {
            try {
                log.debug("task1");
                sleep(1000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }, 1, TimeUnit.SECONDS);

        pool.schedule(() -> {
            log.debug("task2");
            int i = 1 / 0;
        }, 1, TimeUnit.SECONDS);
        pool.schedule(() -> {
            log.debug("task3");
        }, 1, TimeUnit.SECONDS);
    }
}

定时任务每周三执行一次

@Slf4j(topic = "thread.testScheduleTask")
public class TestScheduleTask {
    public static void main(String[] args) {
        //获取当前时间
        LocalDateTime now = LocalDateTime.now();
        //获取周三时间 15:00
        LocalDateTime time = now.withHour(15).withMinute(0).withSecond(0).withNano(0).with(DayOfWeek.WEDNESDAY);
        //如果当前时间大于周三则time加一周
        if (now.compareTo(time) > 0) {
            time = time.plusWeeks(1);
        }
        long initialDelay = Duration.between(now, time).toMillis();
        long period = 1000 * 60 * 60 * 24 * 7;
        ScheduledExecutorService pool = Executors.newScheduledThreadPool(1);
        pool.scheduleAtFixedRate(()->{log.debug("task...");},initialDelay,period, TimeUnit.MILLISECONDS);


    }
}

5.5 工作窃取线程池

内部使用forkJoin进行分治处理任务。

分治：将大任务拆分成算法上相同的小任务，直至不能拆分可以直接求解。

在Fork/Join的基础上加入了多线程，可以把每个任务的分解和合并交给不同的线程池来完成，进一步提升运算效率，它会默认创建和cpu核数相同的线程。

public static ExecutorService newWorkStealingPool() {
        return new ForkJoinPool
            (Runtime.getRuntime().availableProcessors(),
             ForkJoinPool.defaultForkJoinWorkerThreadFactory,
             null, true);
    }

public ForkJoinPool(int parallelism,
                        ForkJoinWorkerThreadFactory factory,
                        UncaughtExceptionHandler handler,
                        boolean asyncMode) {
        this(checkParallelism(parallelism),
             checkFactory(factory),
             handler,
             asyncMode ? FIFO_QUEUE : LIFO_QUEUE,
             "ForkJoinPool-" + nextPoolId() + "-worker-");
        checkPermission();
    }

传入的是线程并发的数量，不传则默认获取系统的cpu核数

该池不保证任务的执行顺序，抢占式的执行

@Slf4j(topic = "thread.testWorkStealingPool")
public class TestWorkStealingPool {
    public static void main(String[] args) throws ExecutionException, InterruptedException {
        ExecutorService pool = Executors.newWorkStealingPool();
        for (int i = 0; i < 16; i++) {

            pool.execute(()-> log.debug("start "));
        }
        Thread.sleep(10000);
        pool.shutdown();
        /*Future<Integer> start = pool.submit(() -> {
            log.debug("start");
            return 1;
        });
        Integer integer = start.get();
        System.out.println(integer);*/

    }
}

6、提交方法

 //提交方法，有返回值 Future获得任务执行结果   
<T> Future<T> submit(Callable<T> task);
//提交tasks中的所有任务，返回泛型为Future的集合
<T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks)
        throws InterruptedException;
//提交tasks中的所有任务,带超时时间
<T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks,
                                  long timeout, TimeUnit unit)
        throws InterruptedException;
//提交tasks中的所有任务，哪个任务先执行完就返回该任务的结果，，其他任务取消
<T> T invokeAny(Collection<? extends Callable<T>> tasks)
        throws InterruptedException, ExecutionException;
//提交tasks中的所有任务，哪个任务先执行完就返回该任务的结果，，其他任务取消，带超时时间
<T> T invokeAny(Collection<? extends Callable<T>> tasks,
                    long timeout, TimeUnit unit)
        throws InterruptedException, ExecutionException, TimeoutException;

6.1 submit

@Slf4j(topic = "thread.TestSubmit")
public class TestSubmit {
    public static void main(String[] args) throws ExecutionException, InterruptedException {
        ExecutorService threadPool = Executors.newFixedThreadPool(2);
        Future<String> future = threadPool.submit(() -> {
            log.debug("running...");
            Thread.sleep(1000);
            return "ok";
        });
        log.debug("{}",future.get());
    }
}

特点：

1、参数为Callable接口的实现类
2、可抛出异常
3、可通过Future对象获取返回值

6.2 invokeAll

private static void testInvokeAll() throws Exception {
        ExecutorService pool = Executors.newFixedThreadPool(2);
        List<Future<String>> futures = pool.invokeAll(Arrays.asList(() -> {
                    log.debug("begin");
                    Thread.sleep(1000);
                    return "1";
                },
                () -> {
                    log.debug("begin");
                    Thread.sleep(500);
                    return "2";
                }, () -> {
                    log.debug("begin");
                    Thread.sleep(2000);
                    return "3";
                }
        ),3500,TimeUnit.MILLISECONDS);
        futures.forEach(f-> {
            try {
                log.debug("{}",f.get());
            } catch (InterruptedException | ExecutionException e) {
                e.printStackTrace();
            }
        });
    }

特点：

1、可以一次提交一个任务集合，并通过Future类型的集合获取返回值

6.3 invokeAny

private static void testInvokeAny() throws InterruptedException, ExecutionException, TimeoutException {
        ExecutorService pool = Executors.newFixedThreadPool(1);
        Object result = pool.invokeAny(Arrays.asList(() -> {
            log.debug("begin 1");
            Thread.sleep(1000);
            return "1";
        }, () -> {
            log.debug("begin 2");
            Thread.sleep(1000);
            return "2";
        }, () -> {
            log.debug("begin 3");
            Thread.sleep(1000);
            return "3";
        }),500, TimeUnit.MILLISECONDS);
        log.debug("{}",result);
    }

7、关闭线程池

shutdown

/*线程池状态变为SHUTDOWN
- 不会接收新任务
- 但已提交任务会执行完
- 中断空闲线程
- 此方法不会阻塞调用线程的执行
*/
void shutdown();
public void shutdown() {
    //获取线程池的全局锁
    final ReentrantLock mainLock = this.mainLock;
    //加锁
        mainLock.lock();
        try {
            //检查是否有关闭线程池的权限
            checkShutdownAccess();
            //修改线程池状态为SHUTDOWN
            advanceRunState(SHUTDOWN);
            //仅会打断空闲线程
            interruptIdleWorkers();
            //为ScheduledThreadPoolExecutor调用钩子函数
            onShutdown(); 
        } finally {
            mainLock.unlock();
        }
      //尝试将状态变为TEMRMINATED
        tryTerminate();
    }
private void checkShutdownAccess() {
        SecurityManager security = System.getSecurityManager();
        if (security != null) {
            security.checkPermission(shutdownPerm);
            final ReentrantLock mainLock = this.mainLock;
            mainLock.lock();
            try {
                for (Worker w : workers)
                    security.checkAccess(w.thread);
            } finally {
                mainLock.unlock();
            }
        }
    }

检查是调用者否有关闭线程池的权限，期间使用了线程池的全局锁。

private void advanceRunState(int targetState) {
        for (;;) {
            int c = ctl.get();
            if (runStateAtLeast(c, targetState) ||
                ctl.compareAndSet(c, ctlOf(targetState, workerCountOf(c))))
                break;
        }
    }

判断当前的线程是否为指定的状态，在shutdow方法中传入的状态是SHUTDOWM，如果是SHUTDOWN，则直接返回，如果不是SHUTDOWN，则将当前线程池的状态设置为SHUTDOWN

 private void interruptIdleWorkers() {
        interruptIdleWorkers(false);
    }
private void interruptIdleWorkers(boolean onlyOne) {
        final ReentrantLock mainLock = this.mainLock;
        mainLock.lock();
        try {
            for (Worker w : workers) {
                Thread t = w.thread;
                if (!t.isInterrupted() && w.tryLock()) {
                    try {
                        t.interrupt();
                    } catch (SecurityException ignore) {
                    } finally {
                        w.unlock();
                    }
                }
                if (onlyOne)
                    break;
            }
        } finally {
            mainLock.unlock();
        }
    }

先获取线程池全局锁，然后遍历所有工作线程，如果它没有被打断并且工作线程获得了锁，则执行线程的中断方法，并且释放线程的锁。此时如果onlyOne参数为true，则跳出循环，最终释放线程池的全局锁。

shutdownNow

shutdownNow方法与shutdown总体逻辑基本相同，只是shutdownNow方法将线程的状态设为STOP，并且中断所有worker线程，同时将任务队列中的剩余任务移动到tasks集合中并返回。

public List<Runnable> shutdownNow() {
        List<Runnable> tasks;
        final ReentrantLock mainLock = this.mainLock;
        mainLock.lock();
        try {
            checkShutdownAccess();
            advanceRunState(STOP);
            interruptWorkers();
            tasks = drainQueue();//剩余任务移到tasks集合中
        } finally {
            mainLock.unlock();
        }
        tryTerminate();
        return tasks;
    }

8、固定线程池面临的问题

newFixedThreadPool newSingleThreadExecutor

饥饿问题（没有线程去执行任务）

@Slf4j(topic = "c.TestHungry")
public class TestHungry {

        static final List<String> result = Arrays.asList("登录成功","登录失败");
        static Random num = new Random();

        static String signIn(){
            return result.get(num.nextInt(result.size()));
        }

    public static void main(String[] args) {
        ExecutorService pool = Executors.newFixedThreadPool(2);
        pool.execute(()->{
            log.debug("注册用户...");
            Future<String> submit = pool.submit(() -> {
                log.debug("登录...");
                return signIn();
            });
            try{
                log.debug("登录结果:{}",submit.get());
            }catch (Exception e){
                e.printStackTrace();
            }
        });

        pool.execute(()->{
            log.debug("注册用户...");
            Future<String> submit = pool.submit(() -> {
                log.debug("登录...");
                return signIn();
            });
            try{
                log.debug("登录结果:{}",submit.get());
            }catch (Exception e){
                e.printStackTrace();
            }
        });

    }
}

解决：

不同的任务使用不同的线程池

9、线程池参数

线程数过小会导致程序不能充分的利用系统资源，容易导致饥饿。
线程数过大会导致更多的线程上下文切换。

9.1 CPU密集型运算

CPU密集的意思是该任务需要大量的运算，而没有阻塞，CPU一直全速运行。CPU密集任务只有在真正的多核CPU上才可能得到加速（通过多线程），而在单核CPU上，无论你开几个模拟的多线程该任务都不可能得到加速，因为CPU总的运算能力就那些。

对于该类任务通常采用cpu核数+1能够实现最优的CPU利用率，+1是保证由于操作系统发生某些故障或者其他原因导致暂停时，额外的线程就能顶上去，保证CPU性能不被浪费。

9.2 I/O密集型运算

CPU 使用率较低，程序中会存在大量的 I/O 操作占用时间，导致线程空余时间很多，所以通常就需要开CPU核心数两倍的线程。当线程进行 I/O 操作 CPU 空闲时，启用其他线程继续使用 CPU，以提高 CPU 的使用率。例如：数据库交互，文件上传下载，网络传输等。

线程数 = 核数 * 期望CPU利用率 * 总时间（CPU计算时间+等待时间）/CPU计算时间

例如四核CPU计算时间为50%，其他等待时间为50%，期望CPU利用率100%

4 * 100% * 100% / 50% = 8

例如四核CPU计算时间为10%，其他等待时间为90%，期望cpu被100%利用

4 * 100% * 100% / 10% = 40

标签：log,debug,任务,线程,------,CPU,pool,小记
From： https://www.cnblogs.com/jamers-rz/p/17700354.html

1、线程池的产生背景

2、线程池

自定义线程池

3、线程池的状态

4、构造函数

5、线程池类别

5.1 newFixedThreadPool

5.2 newCacheThreadPool

5.3 newSingleThreadExecutor

5.4 任务调度线程池

5.5 工作窃取线程池

6、提交方法

6.1 submit

6.2 invokeAll

6.3 invokeAny

7、关闭线程池

8、固定线程池面临的问题

9、线程池参数

9.1 CPU密集型运算

9.2 I/O密集型运算

相关文章

赞助商

阅读排行

线程池------小记

1、线程池的产生背景

2、线程池

自定义线程池

3、线程池的状态

4、构造函数

5、线程池类别

5.1 newFixedThreadPool

5.2 newCacheThreadPool

5.3 newSingleThreadExecutor

5.4 任务调度线程池

5.5 工作窃取线程池

6、提交方法

6.1 submit

6.2 invokeAll

6.3 invokeAny

7、关闭线程池

8、 固定线程池面临的问题

9、线程池参数

9.1 CPU密集型运算

9.2 I/O密集型运算

相关文章

赞助商

阅读排行

　自定义线程池

8、固定线程池面临的问题