Java lambda表达式(二)——java8的CompletableFuture异步并行操作

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20

int size = 5;
CountDownLatch latch = new CountDownLatch(size);
for (int i = 0; i < size; i++) {
    Executors.newFixedThreadPool(size).submit(() -> {
        try {
            // long running task
            System.out.println(Thread.currentThread().getName() + " " + latch.getCount());
        } finally {
            latch.countDown();
        }
    });
}
try {
    latch.await();
} catch (InterruptedException e) {
    e.printStackTrace();
}

// continue...
System.out.println(Thread.currentThread().getName());

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83

import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;

import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;

/**
 * 并行获取各个数据源的数据合并成一个数据组合
 */
public class ParallelTest {
    /**
     * 获取基本信息
     *
     * @return
     */
    public String getProductBaseInfo(String productId) {
        return productId + "商品基础信息";
    }

    /**
     * 获取详情信息
     *
     * @return
     */
    public String getProductDetailInfo(String productId) {
        return productId + "商品详情信息";
    }

    /**
     * 获取sku信息
     *
     * @return
     */
    public String getProductSkuInfo(String productId) {
        return productId + "商品sku信息";
    }

    /**
     * 取得一个商品的所有信息（基础、详情、sku）
     *
     * @param productId
     * @return
     */
    public String getAllInfoByProductId(String productId) {
        CompletableFuture<String> f1 = CompletableFuture.supplyAsync(() -> getProductBaseInfo(productId));
        CompletableFuture<String> f2 = CompletableFuture.supplyAsync(() -> getProductDetailInfo(productId));
        CompletableFuture<String> f3 = CompletableFuture.supplyAsync(() -> getProductSkuInfo(productId));
        //等待三个数据源都返回后，再组装数据。这里会有一个线程阻塞
        CompletableFuture.allOf(f1, f2, f3).join();
        try {
            String baseInfo = f1.get();
            String detailInfo = f2.get();
            String skuInfo = f3.get();
            return baseInfo + "" + detailInfo + skuInfo;
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (ExecutionException e) {
            e.printStackTrace();
        }
        return null;
    }
   /**
    *   并行获取数据的计算
    */
    @Test
    public void testGetAllInfoParalleByProductId() throws ExecutionException, InterruptedException {
        ParallelTest test = new ParallelTest();
        String info = test.getAllInfoByProductId("1111");
        Assertions.assertNotNull(info);
    }
   /**
    *   同步获取执行的处理
    */
    @Test
    public void testGetAllInfoDirectly() throws ExecutionException, InterruptedException {
	 ParallelTest test = new ParallelTest();
        String baseInfo = getProductBaseInfo("1111");
        String detailInfo = getProductDetailInfo("1111");
        String skuInfo = getProductSkuInfo("1111");
        String info=baseInfo + "" + detailInfo + skuInfo;
        Assertions.assertNotNull(info);
    }
}

1

服务器端最佳线程数量=((线程等待时间+线程cpu时间)/线程cpu时间) * cpu数量

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26

public class ParallelTest {
        ExecutorService executor = Executors.newFixedThreadPool(100);    
        /**
         * 取得一个商品的所有信息（基础、详情、sku）
         *
         * @param productId
         * @return
         */
        public String getAllInfoByProductId(String productId) {
            CompletableFuture<String> f1 = CompletableFuture.supplyAsync(() -> getProductBaseInfo(productId),executor);
            CompletableFuture<String> f2 = CompletableFuture.supplyAsync(() -> getProductDetailInfo(productId),executor);
            CompletableFuture<String> f3 = CompletableFuture.supplyAsync(() -> getProductSkuInfo(productId),executor);
    
            try {
                String baseInfo = f1.get();
                String detailInfo = f2.get();
                String skuInfo = f3.get();
                return baseInfo + "" + detailInfo + skuInfo;
            } catch (InterruptedException e) {
                e.printStackTrace();
            } catch (ExecutionException e) {
                e.printStackTrace();
            }
            return null;
        }
    }

1
2
3
4
5

<dependency>
	<groupId>com.twitter</groupId>
	<artifactId>util-core_2.11</artifactId>
	<version>${util-core_2.11.version}</version>
</dependency>