Guava线程池介绍

水深无声 2022-12-16 13:16 120阅读 0赞

## **简介** ##

**       ListenableFuture**顾名思义就是可以监听的**Future**，它是对java原生Future的扩展增强。我们知道Future表示一个异步计算任务，当任务完成时可以得到计算结果。如果我们希望一旦计算完成就拿到结果展示给用户或者做另外的计算，就必须使用另一个线程不断的查询计算状态。这样做，代码复杂，而且效率低下。使用ListenableFuture Guava帮我们检测Future是否完成了，如果完成就自动调用回调函数，这样可以减少并发程序的复杂度。

推荐使用第二种方法，因为第二种方法可以直接得到Future的返回值，或者处理错误情况。本质上第二种方法是通过调动第一种方法实现的，做了进一步的封装

<dependency>
        <groupId>com.google.guava</groupId>
        <artifactId>guava</artifactId>
        <version>19.0</version>
    </dependency>

## 使用方法如下： ##

*  1.创建线程池
 *  2.装饰线程池
 *  3.任务处理
 *  4.回调函数处理
 *  5.所有任务完成后处理

回调方法Futures.addCallback(futureTask,callback,executorService)，FutureCallback操作将会执行在单独的线程，这个线程由传入的ExecutorService参数提供，在这个线程池中进行排队。适合回调处理过程占用CPU高，处理时间较长的情景。

回调过程比较快的则可以考虑另一个回调函数  
**Futures.addCallback(futureTask,callback)**;这样回调函数就是在 ListenableFuture实例执行的线程将上执行FutureCallback操作，即任务将在调用者的线程上运行

## 线程优化的切入点 ##

能进行优化的关键是了解线上线程池的运行状况；所以第一步就是要对线程池进行监控。  
**一个是数据的监控，一个是时间的监控。**

有了记录的参数后，我们要做的是性能瓶颈的分析  
瓶颈可能但不限于出现在一下几个地方：  
**核心线程池数量过小，处理时间过长，导致阻塞队列堆积过大**：可以根据机器性能适当调节核心线程数的大小。同是控制任务提交的评论，能处理多少就提交多少。  
**机器重启导致任务丢失**：控制任务的提交频率，尽量不要排队；或者条件允许，采用可靠的任务提交模式。将未处理完成的任务都记录下来，只有收到处理成功的消息，再删除掉原数据

## 代码实现： ##

private CustomerInfo queryCustomerInfo(String productNo) {
    
        CustomerInfo customerInfo = new CustomerInfo();
        //查询1(泛型为queryOperator方法和queryOperatorCallback方法的返回类型)
        ListenableFuture<Response<String>> operatorInfo = guavaThreadPool.submitAndCallBack(
    		() -> queryOperator(productNo),queryOperatorCallback(customerInfo));
        //查询2
        ListenableFuture carrierInfo= guavaThreadPool.submitAndCallBack(
    		() -> queryCarrier(productNo),queryCarrierCallback(customerInfo));
        // 获取两个线程的执行结果集
        final ListenableFuture listenableFutureResult =
    		Futures.allAsList(operatorInfo , carrierInfo);
        try {
    	// 等待查询1，查询2线程执行成
    	listenableFutureResult.get(1000, TimeUnit.MILLISECONDS);
        } catch (Exception e) {
    	throw new RuntimeException();
        }
        return customerInfo;
    }
    
        /**
         * 提交并添加回调
         *
         * @param callable       线程
         * @param futureCallback 回调函数
         * @return ListenableFuture可监听Future对象
         */
    public <V> ListenableFuture<V> submitAndCallBack(Callable<V> callable, FutureCallback<V> futureCallback) {
        // 1.获取线程池对象
        ThreadPoolExecutor threadPoolExecutor = taskExecutor.getThreadPoolExecutor();
        // 2.使用Guava的ListeningExecutorService装饰线程池
        ListeningExecutorService listeningExecutorService =
                    MoreExecutors.listeningDecorator(threadPoolExecutor);
        // 3.执行任务
        ListenableFuture<V> listenableFuture = listeningExecutorService.submit(callable);
        if (null != futureCallback) {
            // 4.回调处理 回调函数在ListenableFuture实例执行的线程将上执行FutureCallback操作，即任务将在调用者的线程上运行
            Futures.addCallback(listenableFuture, futureCallback);
         }
         return listenableFuture;
    }
    
    
    
    // 查询业务1
    private Response<String> queryOperator(String productNo) {
            Response<String> result = new Response<>();
            try {
                // 具体代码
            } catch (Exception e) {
                // 抛自定义异常
                throw new RuntimeException();
            }
            return result;
    }
    
    
    // 查询业务1的回调方法
    private FutureCallback<Response<String>> queryOperatorCallback(CustomerInfo customerInfo) {
                return new FutureCallback<Response<String>>() {
                @Override
                public void onSuccess(Response<String> response) {
                    if (response != null && response.getResult() != null) {
                        // 成功处理
                    }
                }
                @Override
                public void onFailure(Throwable throwable) {
                    // 失败处理
                }
            };
    }
    
    // 查询业务2
    private Response<String> queryCarrier(String productNo) {
            Response<String> result = new Response<>();
            try {
                // 具体代码
            } catch (Exception e) {
                // 抛自定义异常
                throw new RuntimeException();
            }
            return result;
     }
    
    
    // 查询业务2的回调方法
    private FutureCallback<Response<String>> queryCarrierCallback(CustomerInfo customerInfo) {
            return new FutureCallback<Response<String>>() {
                @Override
                public void onSuccess(Response<String> response) {
                    if (response != null && response.getResult() != null) {
                        // 成功处理
                    }
                }
                @Override
                public void onFailure(Throwable throwable) {
                    // 失败处理
                }
            };
    }