线程池

￡神魔★判官ぃ 2023-05-28 09:56 1阅读 0赞

### 线程池专题 ###

*  循环线程池
 *  常见4种线程池
 *  java线程的生命周期(状态转换)
 *  线程池的执行过程（原理）
 *  线程池实现原理(源码分析)
 *  代码实验证明线程池的运行过程
 *  线程池拒绝策略
 *  线程池demo

# 循环线程池 #

https://blog.csdn.net/u010648555/article/details/78202269  
参考文章  
https://blog.csdn.net/liweizhong193516/article/details/76883571  
scheduleAtFixedRate，不管上次有没有执行完，到了间隔时间就执行（任务开始执行–》开始计算时间）  
scheduleWithFixedDelay等上次执行完后再过多少秒执行（任务执行完–》开始计算时间）

public static void main(String[] args) { 
    	ScheduledExecutorService scheduledThreadPool=Executors.newScheduledThreadPool(3);
    	System.out.println(new Date().getSeconds());
    	scheduledThreadPool.scheduleWithFixedDelay(new Runnable(){ 
    // scheduledThreadPool.scheduleWithFixedDelay(new Runnable(){ 
    		@Override
    		public void run() { 
    			System.out.println(Thread.currentThread().getName()+"***"+new Date().getSeconds()+"===== I love LSR");
    			try { 
    				Thread.sleep(3000);
    			} catch (InterruptedException e) { 
    				// TODO Auto-generated catch block
    				e.printStackTrace();
    			}
    		}
    		//要执行的任务、初始化延时（首次延迟多少秒执行）、线程调用时间间隔（以后每个多少秒执行）、计时单位
    	},1,5,TimeUnit.SECONDS);
    }

public class Abc { 
        public static void main(String[] args) { 
            ScheduledExecutorService uniconf = Executors.newScheduledThreadPool(1,
                    new ThreadFactory() { 
                        @Override
                        public Thread newThread(Runnable runnable) { 
                            Thread thread = new Thread(new ThreadGroup("uniconf"), runnable,
                                    "uniconf" + "-" + "RemoteConfigRepository" + "-" + new AtomicLong(1).getAndIncrement());
                            thread.setDaemon(true);
                            if (thread.getPriority() != Thread.NORM_PRIORITY) { 
                                thread.setPriority(Thread.NORM_PRIORITY);
                            }
                            return thread;
                        }
                    });
            uniconf.submit(new Runnable() { 
                @Override
                public void run() { 
                    System.out.println("哈哈哈哈===>");
                }
            });
            while (true){ 
                try { 
                    TimeUnit.SECONDS.sleep(1);
                } catch (InterruptedException e) { 
                    System.out.println(e);
                }
            } //不加的话，哈哈哈还没来得急打印，主线程都执行完了，
    
        }
    }

# 常见4种线程池 #

创建线程池的几种方式  
Java通过Executors提供四种线程池，分别为：  
newCachedThreadPool创建一个可缓存线程池，如果线程池长度超过处理需要，可灵活回收空闲线程，若无可回收，则新建线程。  
newFixedThreadPool 创建一个定长线程池，可控制线程最大并发数，超出的线程会在队列中等待。  
newScheduledThreadPool 创建一个定长线程池，支**持定时及周期性任务执行**。  
除了线程池的特性以外，可以实现循环或延迟任务。  
newSingleThreadExecutor 创建一个单线程化的线程池，它只会用唯一的工作线程来执行任务，保证所有任务按照指定顺序(FIFO, LIFO, 优先级)执行。

1.  方式1  
    Executors的默认四种线程池本质还是四种不同参数的ThreadPoolExecutor 的示例对象而已，ThreadPoolExecuto最终还是实现的接口Executor

========》
    public class Executors { 
        public static ExecutorService newFixedThreadPool(int nThreads) { 
            return new ThreadPoolExecutor(nThreads, nThreads,
                                          0L, TimeUnit.MILLISECONDS,
                                          new LinkedBlockingQueue<Runnable>());
        }
        public static ScheduledExecutorService newSingleThreadScheduledExecutor() { 
            return new DelegatedScheduledExecutorService
                (new ScheduledThreadPoolExecutor(1));
        }
        。。。。。。。
    }
      ExecutorService  m_executorService =Executors.newCachedThreadPool(ApolloThreadFactory
            .create("ConfigFile", true));
      }

1.  方式2

public class ThreadPoolExecutor extends AbstractExecutorService { }
    public abstract class AbstractExecutorService implements ExecutorService { }
    public interface ExecutorService extends Executor { }
    
    ThreadPoolExecutor  executor = new ThreadPoolExecutor(10, 20, 1,
                    TimeUnit.SECONDS, new ArrayBlockingQueue<>(10), Executors.defaultThreadFactory(), new ThreadPoolExecutor.DiscardPolicy());

# java线程的生命周期(状态转换) #

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQxMDYzMTQx_size_16_color_FFFFFF_t_70]  
[原文][Link 1]

# 线程池的执行过程（原理） #

> 初始最小线程10个，最大线程20个，任务队列容量为10个，线程池启动，此时没有任何线程，现在有10个任务过来了，线程池会创建10个线程(即核心线程)去执行这10个任务，10个线程正在执行这个任务时，又来了10个任务，此时会将这10个任务放到任务队列里面，此时又来了10个任务，线程池会创建10个非核心线程去执行这10个任务，此时如果再来新的任务，就会采取线程池的拒绝策略。\[当前有20个线程在执行20个任务，还有10个任务位于任务队列(又叫阻塞队列)\]，然后在这20个正在运行的线程当中，先执行完的10个线程就会去任务队列里面依次消费这10个任务，后执行玩的10个线程，再去任务队列里面取任务时，发现任务队列为空，而当前活跃线程为20，超过了核心线程，所以会将这10个线程一一销毁,当剩余的10个线程执行完任务队列中的任务时，会再次去任务队列中获取任务，此时任务队列为空且线程数小于或等于核心线程数，此时这十个线程将一一进入为阻塞状态。直到任务队列中有新的任务来时，才会将线程唤醒，去执行任务队列中的任务  
> ————————————————  
> [原文链接][Link 2]

# 线程池实现原理(源码分析) #

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQxMDYzMTQx_size_16_color_FFFFFF_t_70 1]

> 一个线程集合workerSet和一个阻塞队列workQueue。当用户向线程池提交一个任务时，线程池将任务放入workQueue中。workerSet中的线程会不断的从workQueue中获取任务然后执行。当workQueue中没有任务的时候，worker就会阻塞，直到队列中有任务了就取出来继续执行。  
> ————————————————

//1. 创建线程池====new ThreadPoolExecutor
    ThreadPoolExecutor pool = new ThreadPoolExecutor(2, 3, 60L, TimeUnit.SECONDS,new LinkedBlockingQueue<>(1));
           //2. 用户向线程池提交一个任务====pool.execute
            pool.execute(new Runnable() { 
    
                @Override
                public void run() { 
                    System.out.println("-------------helloworld_001---------------" + Thread.currentThread().getName());
                }
            });
     //进入pool.execute（）方法
     //工作线程《核心线程
     if (workerCountOf(c) < corePoolSize) { 
     //直接创建new一个线程去执行任务
                if (addWorker(command, true)
    
    //====》进入
                Worker w = new Worker(firstTask);
                /** * private final class Worker extends AbstractQueuedSynchronizer implements Runnable */
                /** *private final HashSet<Worker> workers = new HashSet<Worker>(); */
                workers.add(w);
                //执行
                public void run() { 
                runWorker(this); //===》进入
                Runnable task = w.firstTask;
                 task.run();  
            }
            //====》活跃线程数》核心线程数
            workQueue.offer(command)
            //以arrayblockqueue为例
            public boolean offer(E e) { 
            checkNotNull(e);
            final ReentrantLock lock = this.lock;
            lock.lock();
            try { 
                if (count == items.length)
                    return false;
                else { 
                    enqueue(e);//======》
                    return true;
                }
            } finally { 
                lock.unlock();
            }
        }
    //====》
     final void runWorker(Worker w) { 
         while (task != null || (task = getTask()) != null) { 
                    w.lock();
                    try{ task.run();}
    //===》进入getTask()
    //刚开始，队列为空
    if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) { 
                    decrementWorkerCount();
                    return null;
                }
    if（workerCountOf(c)》corePoolSize）{ 
    Runnable r = timed ?
                        workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :
                        /** *超过核心线程数，并且空闲线程超过空闲等待时间，就销毁多余的线程 * */
                        workQueue.take();  //没有超过核心线程，就阻塞核心线程
                          //while (count == 0)
                    //notEmpty.await(); 线程阻塞
                //return dequeue();
               /** *关于Condition参考 https://www.jianshu.com/p/481395a28322 * private final Condition notEmpty; */
               
              public E poll(long timeout, TimeUnit unit) throws InterruptedException { 
            long nanos = unit.toNanos(timeout);
            final ReentrantLock lock = this.lock;
            lock.lockInterruptibly();
            try { 
                while (count == 0) { 
                    if (nanos <= 0)  //超过空闲等待后线程还没有被唤醒，就返回null，即销毁线程
                        return null;
                    nanos = notEmpty.awaitNanos(nanos); //线程阻塞一段时间（设置得空闲等待时间）
                }
                return dequeue();
            } finally { 
                lock.unlock();
            }
        }
    }

[原文链接][Link 3]

# 代码实验证明线程池的运行过程 #

[原文链接][Link 4]

# 线程池拒绝策略 #

[原文链接][Link 5]  
AbortPolicy:丢弃任务并抛出RejectedExecutionException异常。 【关键业务中使用，在系统不能承载更大的并发量的时候，能够及时的通过异常发现】【线程池默认的拒绝策略】（流产-中止）  
DiscardPolicy：丢弃任务，但是不抛出异常。（丢弃）  
DiscardOldestPolicy：丢弃队列最前面的任务，然后重新提交被拒绝的任务 CallerRunsPolicy：由调用线程（提交任务的线程）处理该任务

# 线程池demo #

ThreadPoolExecutor  executor = new ThreadPoolExecutor(10, 20, 1,
                    TimeUnit.SECONDS, new ArrayBlockingQueue<>(10), Executors.defaultThreadFactory(), new ThreadPoolExecutor.DiscardPolicy());
            for (int i = 0; i <5 ; i++) { 
                executor.execute(()->{ 
                    try { 
                        TimeUnit.SECONDS.sleep(3);
                        System.out.println("睡了3秒");
                    } catch (InterruptedException e) { 
                        e.printStackTrace();
                    }
                });
            }

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQxMDYzMTQx_size_16_color_FFFFFF_t_70]: /images/20230528/e608931c7f764fa89c0580de4387e7e3.png
[Link 1]: https://www.cnblogs.com/marsitman/p/11228684.html
[Link 2]: https://www.cnblogs.com/DDiamondd/p/11362164.html
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQxMDYzMTQx_size_16_color_FFFFFF_t_70 1]: /images/20230528/e85501937e5748edbbb7cd87a6d29194.png
[Link 3]: https://blog.csdn.net/u013332124/article/details/79587436
[Link 4]: https://www.cnblogs.com/cdf-opensource-007/p/8769777.html
[Link 5]: https://blog.csdn.net/suifeng629/article/details/98884972