java自定义线程池ThreadPoolExecutor的创建和线程池在项目中的实际使用

布满荆棘的人生 2022-11-05 00:45 275阅读 0赞

### 为什么用线程池 ###

线程池做的主要工作就是控制运行的线程的数量，处理过程中，将任务放入到队列中，然后线程创建后，启动这些任务，如果线程数量超过了最大数量的线程排队等候，等其它线程执行完毕，再从队列中取出任务来执行。它的主要特点为：线程复用、控制最大并发数、管理线程

### 最常见的线程池 ###

// 创建一个定长线程池，可控制线程最大并发数，超出的线程会在队列中等待。
            ExecutorService threadPool = Executors.newFixedThreadPool(5);
    
             // 创建一个单线程化的线程池，它只会用唯一的工作线程来执行任务，保证所有任务按照指定顺序(FIFO, LIFO, 优先级)执行。
            ExecutorService threadPoo2 = Executors.newSingleThreadExecutor();
    
            // 创建一个可缓存线程池，如果线程池长度超过处理需要，可灵活回收空闲线程，若无可回收，则新建线程
            ExecutorService threadPoo3 = Executors.newCachedThreadPool();
    
            // 创建一个定长线程池，支持定时及周期性任务执行。
            ExecutorService threadPoo4 = Executors.newScheduledThreadPool(5);

### 线程池七大核心参数 ###

public ThreadPoolExecutor(int corePoolSize,
                                  int maximumPoolSize,
                                  long keepAliveTime,
                                  TimeUnit unit,
                                  BlockingQueue<Runnable> workQueue) {
            this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,
                 Executors.defaultThreadFactory(), defaultHandler);
        }

核心参数的判断逻辑

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**：核心线程数，线程池中的常驻核心线程数

**maximumPoolSize**：线程池能够容纳同时执行的最大线程数，此值必须大于等于1

**keepAliveTime**：多余的空闲线程存活时间

**unit**：keepAliveTime的单位

**workQueue**：任务队列，被提交的但未被执行的任务（类似于银行里面的候客区）

**threadFactory**：表示生成线程池中工作线程的线程工厂，用于创建线程池 一般用默认即可

**handler**：拒绝策略，表示当队列满了并且工作线程大于线程池的最大线程数（maximumPoolSize3）时，如何来拒绝请求执行的Runnable的策略

### 线程池处理任务流程说明 ###

1.  在创建了线程池后，等待提交过来的任务请求
2.  当调用execute()方法添加一个请求任务时，线程池会做出如下判断
    
    1.  如果正在运行的线程池数量小于corePoolSize，那么马上创建线程运行这个任务
    2.  如果正在运行的线程数量大于或等于corePoolSize，那么将这个任务放入队列
    3.  如果这时候队列满了，并且正在运行的线程数量还小于maximumPoolSize，那么还是创建非核心线程like运行这个任务；
    4.  如果队列满了并且正在运行的线程数量大于或等于maximumPoolSize，那么线程池会启动饱和拒绝策略来执行
3.  当一个线程完成任务时，它会从队列中取下一个任务来执行
4.  当一个线程无事可做操作一定的时间(keepAliveTime)时，线程池会判断：
    
    1.  如果当前运行的线程数大于corePoolSize，那么这个线程就被停掉
    2.  所以线程池的所有任务完成后，它会最终收缩到corePoolSize的大小

**（ maximumPoolSize +  workQueue ）= 单次最大线程数，超出会被拒绝 ，5+3 = 8 ，所以这里只输出了8次**

我们看一下测试

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2h1cWlhbmt1bmxvbA_size_16_color_FFFFFF_t_70][]

首先我们手动创建线程池，首先是处理层，我这里为了简单，直接写在controller里面

/**
     * @version V1.0
     * @author: hqk
     * @date: 2021/3/3 10:29
     * @Description: 多线程分页处理数据测试
     */
    @RestController
    public class TestController {
    
        @Autowired
        private OrderInfoMapper orderInfoMapper;
    
        /**
         * 多线程分页处理数据
         * @return
         */
        @RequestMapping("ceshi")
        public AjaxResult createOrderinfo(){
            
            ThreadPoolExecutor pool = new ThreadPoolExecutor(3,3,60,
                    TimeUnit.SECONDS,new LinkedBlockingDeque<Runnable>(10),
                    Executors.defaultThreadFactory(),
                    new ThreadPoolExecutor.DiscardOldestPolicy());
    
            PageHelper.startPage(1,10);
            List<Map<String,String>> list = orderInfoMapper.selectList();
            PageInfo<Map<String,String>> pageInfo = new PageInfo<>(list);
            // 获取一共有多少页
            int pages = pageInfo.getPages();
    
            for (int i = 1; i <= pages ; i++) {
                final int taskId = i;
                pool.execute(new MyTask(taskId,orderInfoMapper));
            }
    
          return AjaxResult.success();
        }
    }

**首先获取需要分页查询一共有多少页，关于核心线程池数和最大线程数，一般可设置CPU核数+1，（CPU密集型） ，CPU核数 \* 2 （IO密集型）,如果总页数大于最大线程数，**

**队列数可以用总页数减最大线程数**

然后看下任务线程

/**
     * @version V1.0
     * @author: hqk
     * @date: 2021/3/3 14:26
     * @Description: 线程处理数据
     */
    public class MyTask implements Runnable {
    
        private int pageNum;
    
        private OrderInfoMapper orderInfoMapper;
    
        public int getTaskId() {
            return pageNum;
        }
    
        public void setTaskId(int pageNum) {
            this.pageNum = pageNum;
        }
    
    
        public MyTask(int pageNum, OrderInfoMapper orderInfoMapper) {
            this.pageNum = pageNum;
            this.orderInfoMapper = orderInfoMapper;
        }
    
    
        @Override
        public void run() {
            try {
    
                PageHelper.startPage(pageNum,10);
                List<Map<String,String>>  list = orderInfoMapper.selectList();
    
                if (list!=null&&list.size()>0){
                    for (int i=0;i<list.size();i++){
                        System.out.println(Thread.currentThread().getName()+"--pageNum:" + pageNum+"---"+list.get(i));
                    }
                }else {
                    System.out.println("==========================================");
                }
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    
    }

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2h1cWlhbmt1bmxvbA_size_16_color_FFFFFF_t_70 1][]

这里主要代码已经测试完毕，经过测试不会存在重复读取数据。没有贴出来的就是查询数据库，如果需要的话可留言

我们在实际开发中的哪些场景可以使用线程池呢，比如批量处理/发送 数据，每个线程处理两千条这种

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2h1cWlhbmt1bmxvbA_size_16_color_FFFFFF_t_70]: /images/20221023/8f28ad1a48eb413a832dff51b2e931a3.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2h1cWlhbmt1bmxvbA_size_16_color_FFFFFF_t_70 1]: /images/20221023/f2f27bfe3c7f4cefa495fd820d8f463a.png