java线程池实战

线程池是一种常用的并发编程技术，它可以有效管理和复用线程，提高程序的性能和资源利用率。线程池本质是池化技术，和连接池类似，创建连接与关闭连接属于耗时操作，创建线程与销毁线程也属于耗时重操作。为了提高效率，先提前创建好一批线程，当有需要使用线程时从线程池取出，用完后放回线程池，这样避免了频繁创建与销毁线程。

线程池参数

核心参数

线程池的核心参数决定了池的类型，进而决定了池的特性。

参数与池的关系

Executors类默认创建线程池与参数对应关系。 

线程池类型

根据使用场景选择对应的线程池。

提交任务方式

往线程池中提交任务，主要有两种方法：提交无返回值的任务和提交有返回值的任务。

• execute用于提交不需要返回结果的任务。
• submit用于提交一个需要返回果的任务。该方法返回一个Future对象，通过调用这个对象的get()方法，我们就能获得返回结果。get()方法会一直阻塞，直到返回结果返回。

 在提交任务时，如果无返回值任务，优先使用execute方法。

关闭线程池

在线程池使用完成之后，我们需要对线程池中的资源进行释放操作，这就涉及到关闭功能。

1.关闭线程池的意义 

线程池作为任务的管理者，需要优雅的关闭，原因是如果不手动关闭线程池，线程池中正在执行执行的线程以及队列中还未执行任务将会变得极不可控，如果处理不当，可能造成数据丢失，业务请求结果不正确等问题。

如果业务能够忽略上述影响，那么直接关闭JVM，那么连接池资源会自动释放，不需要手动关闭。

 2.关闭线程池的方式

调用线程池对象的shutdown()和shutdownNow()方法来关闭线程池。

• shutdown()会将线程池状态置为SHUTDOWN，不再接受新的任务，同时会等待线程池中已有的任务执行完成再结束。
• shutdownNow()会将线程池状态置为SHUTDOWN，对所有线程执行interrupt()操作，清空队列，并将队列中的任务返回回来。

关闭线程池涉及到两个返回boolean的方法，isShutdown()和isTerminated，分别表示是否关闭和是否终止。 

工作队列

工作队列是BlockingQueue类型的，理论上只要是它的子类，我们都可以用来作为等待队列。常用的工作队列有四种。

• ArrayBlockingQueue 队列是有界的，基于数组实现的阻塞队列
• LinkedBlockingQueue 队列可以有界，也可以无界。基于链表实现的阻塞队列
• SynchronousQueue 不存储元素的阻塞队列，每个插入操作必须等到另一个线程调用移除操作，否则插入操作将一直处于阻塞状态。该队列是Executors.newCachedThreadPool()的默认队列
• PriorityBlockingQueue 带优先级的无界阻塞队列

通常情况下，我们需要指定阻塞队列的上界（比如1024）。如果执行的任务很多，我们可能需要将任务进行分类，然后将不同分类的任务放到不同的线程池中执行。

拒绝策略

所谓拒绝策略，是指当线程池满了、队列也满了的时候，对新提交任务采取的态度。常见如下四种策略：

• CallerRunsPolicy 在调用者线程执行
• AbortPolicy 直接抛出RejectedExecutionException异常
• DiscardPolicy 任务直接丢弃，不做任何处理
• DiscardOldestPolicy 丢弃队列里最旧的那个任务，再尝试执行当前任务

这四种策略各有优劣，比较常用的是DiscardPolicy。如果需要实现自定义的拒绝策略， 通过实现RejectedExecutionHandler接口的方式来实现。

创建线程池

 /**
     * 核心线程池数量
     */
    int corePoolSize = 5;
    /**
     * 最大线程池的数量 非核心线程池=最大线程池-核心线程池
     */
    int maximumPoolSize = 10;
    /**
     * 非核心线程池最大空闲时间
     */
    int keepAliveTime = 1;
    /**
     * 非核心线程池最大空闲时间单位
     */
    TimeUnit unit = TimeUnit.MINUTES;
    /**
     * 阻塞队列
     */
    BlockingQueue workQueue = new ArrayBlockingQueue<>(10);
    /**
     * 线程池工厂
     */
    ThreadFactory threadFactory = Executors.defaultThreadFactory();
    /**
     * 异常处理策略（报错）
     */
    RejectedExecutionHandler handler = new ThreadPoolExecutor.AbortPolicy();
    /**
     * 手动构造线程池
     */
    ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, threadFactory, handler);
    threadPoolExecutor.execute(() ->{
        System.out.println("开始任务");
    });

实战

现有一个商品地区关联表，地区表，商品表，我需要从商品地区关联表获取商品id，再从商品id获取商品表的商品信息，并顺便从商品信息从获取卖家信息。

public ResponseResult selectService(SelectServiceDTO selectServiceDTO) {
        // 获取对应地区的商品id
        List serviceCampusInfos = serviceInfoMapper.selectServiceByCapmpusId(selectServiceDTO.getCampusId(), (selectServiceDTO.getPage() - 1) * selectServiceDTO.getLimit(), selectServiceDTO.getLimit());
 
        if (serviceCampusInfos.isEmpty()) {
            return new ResponseResult(HttpCodeEnum.SUCCESS.getCode(), null);
        }
        // 返回服务校区信息数量和当前系统可用处理器数量乘以2中较小的那个值。
        int threadNum = Math.min(serviceCampusInfos.size(), Runtime.getRuntime().availableProcessors() * 2);
        ExecutorService executorService = new ThreadPoolExecutor(
                threadNum,
                threadNum,
                60L,
                TimeUnit.SECONDS,
                new LinkedBlockingQueue<>(),
                Executors.defaultThreadFactory(),
                new ThreadPoolExecutor.AbortPolicy()
        );
 
        List> futures = serviceCampusInfos.stream()
                .map(serviceCampusInfo -> CompletableFuture.supplyAsync(() -> {
                    // 获取商品信息
                    ServiceInfo serviceInfo = serviceInfoMapper.selectOne(new LambdaQueryWrapper()
                            .eq(ServiceInfo::getId, serviceCampusInfo.getServiceId()));
                    // 获取商品图片
                    List imageIdList = JSON.parseArray(serviceInfo.getServiceImgIdList(), Integer.class);
                    List imageUrlList = imageIdList.stream()
                            .map(imageInfoMapper::selectImageUrlById)
                            .collect(Collectors.toList());
                    serviceInfo.setServiceImgUrlList(imageUrlList);
                    // 获取对应商品的师傅信息
                    MasterInfo masterInfo = masterInfoMapper.selectOne(new LambdaQueryWrapper()
                            .eq(MasterInfo::getId, serviceInfo.getServiceMasterId()));
                    // 获取师傅图片
                    masterInfo.setMasterImageUrl(imageInfoMapper.selectImageUrlById(masterInfo.getMasterImageId()));
 
                    serviceInfo.setMasterInfo(masterInfo);
 
                    return serviceInfo;
                }, executorService))
                .collect(Collectors.toList());
 
        CompletableFuture allFutures = CompletableFuture.allOf(futures.toArray(new CompletableFuture[0]));
 
        try {
            allFutures.get();
        } catch (InterruptedException | ExecutionException e) {
            // 处理异常
        }
 
        List serviceInfoVOS = futures.stream()
                .map(CompletableFuture::join)
                .map(this::convertToServiceInfoVO)
                .collect(Collectors.toList());
 
        executorService.shutdown();
 
        return new ResponseResult(HttpCodeEnum.SUCCESS.getCode(), serviceInfoVOS);
    }
 
    private ServiceInfoVO convertToServiceInfoVO(ServiceInfo serviceInfo) {
        ServiceInfoVO serviceInfoVO = new ServiceInfoVO();
        BeanUtils.copyProperties(serviceInfo, serviceInfoVO);
 
        MasterInfoVO masterInfoVO = new MasterInfoVO();
        BeanUtils.copyProperties(serviceInfo.getMasterInfo(), masterInfoVO);
 
        serviceInfoVO.setMasterInfoVO(masterInfoVO);
 
        return serviceInfoVO;
    }