手写Java线程池

﹏ヽ暗。殇╰゛Y 2023-05-31 15:29 14阅读 0赞

**先看一下阻塞队列**

*  1、ArrayBlockingQueue：是一个基于数组结构的有界阻塞队列，此队列按 FIFO（先进先出）原则对元素进行排序。
 *  2、LinkedBlockingQueue：一个基于链表结构的阻塞队列，此队列按FIFO （先进先出） 排序元素，吞吐量通常要高于ArrayBlockingQueue。静态工厂方法Executors.newFixedThreadPool()使用了这个队列。
 *  3、SynchronousQueue：一个不存储元素的阻塞队列。每个插入操作必须等到另一个线程调用移除操作，否则插入操作一直处于阻塞状态，吞吐量通常要高于LinkedBlockingQueue，静态工厂方法Executors.newCachedThreadPool使用了这个队列。
 *  4、PriorityBlockingQueue：一个具有优先级得无限阻塞队列。  
    threadFactory：用于设置创建线程的工厂，可以通过线程工程给每个创建出来的线程设置更有意义的名字。  
    handler：饱和策略，当线程池和队列都满了，说明线程池处于饱和状态，那么必须采取一种策略处理提交的新任务。这个策略默认情况下是AbortPolicy，表示无法处理新任务时抛出异常。

RejectedExecutionException:
         1、AbortPolicy策略：该策略会直接抛出异常，阻止系统正常工作；
          
          2、CallerRunsPolicy策略：如果线程池的线程数量达到上限，该策略会把任务队列中的任务放在调用者线程当中运行；
          
          3、DiscardOledestPolicy策略：该策略会丢弃任务队列中最老的一个任务，也就是当前任务队列中最先被添加进去的，马上要被执行的那个任务，并尝试再次提交；
          
          4、DiscardPolicy策略：该策略会默默丢弃无法处理的任务，不予任何处理。当然使用此策略，业务场景中需允许任务的丢失；

此外，我们还可以通过调用Ececutors中的某个静态工厂方法来创建一个线程池(它们的内部实现原理都是相同的，仅仅是使用了不同的工作队列或线程池大小)：

可缓存线程池，先查看池中有没有以前建立的线程，如果有，就直接使用。如果没有，就建一个新的线程加入池中，缓存型池子通常用于执行一些生存期很短的异步型任务。
        线程池为无限大，当执行当前任务时上一个任务已经完成，会复用执行上一个任务的线程，而不用每次新建线程
        ExecutorService service = Executors.newCachedThreadPool();
        
        
        创建一个可重用固定个数的线程池，以共享的无界队列方式来运行这些线程。
        ExecutorService service1 = Executors.newFixedThreadPool(3);
        
        
        定长线程池的大小最好根据系统资源进行设置
        int i1 = Runtime.getRuntime().availableProcessors();
        
        
        创建一个定长线程池，支持定时及周期性任务执行
        ScheduledExecutorService service2 = Executors.newScheduledThreadPool(3);
        scheduledThreadPool.schedule();
        scheduledThreadPool.scheduleAtFixedRate();
        
        
        创建一个单线程化的线程池，它只会用唯一的工作线程来执行任务，保证所有任务按照指定顺序(FIFO, LIFO, 优先级)执行。
        ExecutorService service3 = Executors.newSingleThreadExecutor();

**手写线程池**  
实现线程池最大线程数个数，任务太多就排队，队列满了就不接收任务了。

import java.util.ArrayList;
    import java.util.concurrent.ArrayBlockingQueue;
    import java.util.concurrent.BlockingQueue;
    import java.util.concurrent.locks.ReentrantLock;
    
    public class TestThreadPoolExecutos { 
    
        //存放线程的集合
        private final ArrayList<Mthread> mthreads;
    
        //存放任务队列
        private final BlockingQueue<Runnable> taskQueue;
    
        //线程池限定大小
        private final int threadNum;
    
        //已经工作的线程池数目
        private int workThreadNum;
    
        //是否关闭线程池
        private volatile boolean isShutDown;
    
        //锁
        private final ReentrantLock mainLock = new ReentrantLock();
    
    
        public TestThreadPoolExecutos(int threadNum) { 
            mthreads = new ArrayList<>(threadNum);
            taskQueue = new ArrayBlockingQueue<>(threadNum);
            this.threadNum = threadNum;
            this.workThreadNum = 0;
        }
    
        void execute(Runnable runnable) { 
            if (isShutDown) { 
                throw new RuntimeException("线程池已经关闭");
            }
            try { 
                mainLock.lock();
                //如果工作线程小于总线程数，就开启线程
                if (workThreadNum < threadNum) { 
                    Mthread thread = new Mthread(runnable);
                    thread.start();
                    mthreads.add(thread);//防止gc回收掉
                    workThreadNum++;
                }else { 
                    //添加到任务队列
                    if (!taskQueue.offer(runnable)){ 
                        System.out.println("队列已满");
                    }
                }
    
            } finally { 
                mainLock.unlock();
            }
        }
    
        void shutDown(){ 
            for (Mthread mthread : mthreads) { 
                mthread.interrupt();
            }
            isShutDown=true;
        }
    
        class Mthread extends Thread { 
    
            Runnable task;
    
            public Mthread(Runnable task) { 
                this.task = task;
            }
    
            @Override
            public void run() { 
                while (!Thread.currentThread().isInterrupted()){ 
                    if (task!=null){ 
                        task.run();
                        task=null;
                    }else { 
                        //当前线程处理完自己的了，就会处理队列中的任务
                        Runnable poll = taskQueue.poll();
                        if (poll!=null){ 
                            poll.run();
                        }
                    }
                }
            }
        }
    
        public static void main(String[] args) { 
            TestThreadPoolExecutos testThreadPoolExecutos = new TestThreadPoolExecutos(10);
            for (int i = 0; i < 20; i++) { 
                testThreadPoolExecutos.execute(()->{ 
                    System.out.println(Thread.currentThread().getName()+" is running");
                });
            }
        }
    }

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