ThreadPoolExecutor配合CompletableFuture提高并发处理

关于CompletableFuture的介绍参考博文：

对线程池ThreadPoolExecutor有多种选择，如：ThreadPoolExecutor、newFixedThreadPool()、newSingleThreadExecutor()、newCachedThreadPool()、newScheduledThreadPool。对线程池的详细了解参考：

CompleteableFuture基于jdk8。提高程序吞吐量，异步处理请求。

示例为对一个对象的输出，实现如下。

实体类及其正常输出与异步输出耗时对比

package threads.completableFutureAndExecutor.noThreadPool;import java.util.Random;import java.util.concurrent.CompletableFuture;import java.util.concurrent.Future;public class Shop {
       Random random = new Random();    private String name;    public String getName() {
           return name;    }    //constructor    public Shop(String name) {
           super();        this.name = name;    }    public static void delay() {
           try {
               Thread.sleep(1000L);        } catch (Exception e) {
               throw new RuntimeException(e);        }    }    public double getPrice() {
           delay();        return random.nextDouble() * 100;    }    /**     * 1.将getPrice()方法转为异步调用。     * 2.supplyAsync参数是一个生产者。Supplier 没有参数返回一个对象。若lambda没有参数并且可以返回对象则满足     * 3.supplyAsync内部对Supplier方法对异常进行处理，避免因为异常程序永久阻塞。     * */    public Future
   
     getPriceAsync(String product){
           return CompletableFuture.supplyAsync(()->getPrice());    }    /*    *异步优势示例    */    public static void main(String[] args) throws Exception{
           Shop shop = new Shop("bestShop");                long start = System.currentTimeMillis();        Future
    
      futurePrice = shop.getPriceAsync("some product");        System.out.println("异步调用返回，耗时"+(System.currentTimeMillis() - start));        double price = futurePrice.get();        System.out.println("非异步返回，耗时"+(System.currentTimeMillis() - start));    }}

输出结果：

异步调用返回，耗时73非异步返回，耗时1073

多线程实现：并行与串行对比

package threads.completableFutureAndExecutor.threadPool;import threads.completableFutureAndExecutor.noThreadPool.Shop;import java.util.Arrays;import java.util.List;import java.util.concurrent.*;import java.util.stream.Collectors;public class Price {
       private List
   
     shops = Arrays.asList(new Shop("shop1"),            new Shop("shop2"),            new Shop("shop3"),            new Shop("shop4"),            new Shop("shop5"),            new Shop("shop6"),            new Shop("shop7"),            new Shop("shop8"),            new Shop("shop9"),            new Shop("shop10"),            new Shop("shop12"),            new Shop("shop13"),            new Shop("shop13"),            new Shop("shop14"),            new Shop("shop15"),            new Shop("shop16"),            new Shop("shop17"),            new Shop("shop18"));    //普通方式    public List
    
      findPrices(){
           return shops.stream().parallel().map(shop->                String.format("%s price is %.2f", shop.getName(), shop.getPrice()))                .collect(Collectors.toList());    }    //串行执行	public List
     
       findPriceByJoin(){
   		List
      
       
        > priceFuture = shops.stream()				.map(shop -> CompletableFuture						.supplyAsync(()-> String.format("%s price is %.2f", shop.getName(), shop.getPrice())))				.collect(Collectors.toList());		return priceFuture.stream().map(CompletableFuture::join).collect(Collectors.toList());//串行执行	}	//并行执行    public List
        
          findPriceExecutorsCompletableFuture(){ ExecutorService executor = Executors.newFixedThreadPool(Math.min(shops.size(), 100)); List
         
          
           > priceFuture = shops.stream() .map(shop -> CompletableFuture.supplyAsync(() -> String.format("%s price is %.2f", shop.getName(), shop.getPrice()), executor)) .collect(Collectors.toList()); try { return priceFuture.parallelStream().map(completableFuture -> { String result = ""; try { result = completableFuture.get(2, TimeUnit.SECONDS); } catch (InterruptedException e) { e.printStackTrace(); } catch (ExecutionException e) { e.printStackTrace(); } catch (TimeoutException e) { e.printStackTrace(); } return result; }).collect(Collectors.toList()); } catch (Exception e) { e.printStackTrace(); }finally { //Executors作为局部变量时，创建了线程，一定要记得调用executor.shutdown();来关闭线程池，如果不关闭，会有线程泄漏问题。 executor.shutdown(); } return null; } public static void main(String[] args) { Price priceDemo = new Price(); long start = System.currentTimeMillis(); System.out.println(priceDemo.findPrices()); System.out.println("服务耗时："+(System.currentTimeMillis()-start)); start = System.currentTimeMillis(); System.out.println(priceDemo.findPriceByJoin() ); System.out.println("服务耗时："+(System.currentTimeMillis()-start)); start = System.currentTimeMillis(); System.out.println(priceDemo.findPriceExecutorsCompletableFuture()); System.out.println("服务耗时："+(System.currentTimeMillis()-start)); }}

输出结果：

[shop1 price is 46.71, shop2 price is 80.96, shop3 price is 68.59, shop4 price is 49.26, shop5 price is 40.12, shop6 price is 39.93, shop7 price is 73.94, shop8 price is 17.58, shop9 price is 80.07, shop10 price is 45.05, shop12 price is 34.72, shop13 price is 2.41, shop13 price is 9.34, shop14 price is 76.79, shop15 price is 56.85, shop16 price is 83.44, shop17 price is 19.26, shop18 price is 73.76]服务耗时：3104[shop1 price is 31.64, shop2 price is 22.38, shop3 price is 21.67, shop4 price is 37.91, shop5 price is 43.82, shop6 price is 35.10, shop7 price is 94.76, shop8 price is 13.75, shop9 price is 20.29, shop10 price is 65.15, shop12 price is 43.14, shop13 price is 62.76, shop13 price is 81.70, shop14 price is 19.87, shop15 price is 68.99, shop16 price is 48.44, shop17 price is 71.09, shop18 price is 37.61]服务耗时：3004[shop1 price is 30.43, shop2 price is 97.64, shop3 price is 79.26, shop4 price is 52.21, shop5 price is 52.05, shop6 price is 81.84, shop7 price is 51.62, shop8 price is 29.53, shop9 price is 94.95, shop10 price is 79.17, shop12 price is 0.41, shop13 price is 16.56, shop13 price is 73.60, shop14 price is 80.24, shop15 price is 1.34, shop16 price is 51.02, shop17 price is 0.10, shop18 price is 71.72]服务耗时：1007

结果一看就很明了，并行实现异步输出18个记录比串行异步实现节省了2/3的时间，另外，并行实现异步输出18个记录与正常输出一个记录的耗时是基本一样的。这对执行多个任务时很有帮助。

参考链接：

转载地址：http://wzden.baihongyu.com/

你可能感兴趣的文章