CompletableFuture集合threadPool来提高并发处理
ThreadPool
A simple C++11 Thread Pool implementation
项目地址:https://gitcode.com/gh_mirrors/th/ThreadPool
·
CompleteableFuture基于jdk8。提高程序吞吐量,异步处理请求。
public class Shop {
Random random = new Random();
private String name;
public String getName() {
return name;
}
public void setName(String name) {
this.name = name;
}
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(String product) {
delay();
return random.nextDouble() * 100;
}
// public Future<Double> getPriceAsync(String product){
// CompletableFuture<Double> futurePrice = new CompletableFuture<>();
// new Thread(()->futurePrice.complete(getPrice(product))).start();
// return futurePrice;
// }
/**推荐使用这种方式
* 1.将getPrice()方法转为异步调用。
* 2.supplyAsync参数是一个生产者。Supplier 没有参数返回一个对象。若lambda没有参数并且可以返回对象则满足
* 3.supplyAsync内部对Supplier方法对异常进行处理,避免因为异常程序永久阻塞。
* */
public Future<Double> getPriceAsync(String product){
return CompletableFuture.supplyAsync(()->getPrice(product));
}
public static void main(String[] args) throws Exception{
Shop shop = new Shop("bestShop");
long start = System.currentTimeMillis();
Future<Double> futurePrice = shop.getPriceAsync("some product");
System.out.println("调用返回,耗时"+(System.currentTimeMillis() - start));
double price = futurePrice.get();
System.out.println("价格返回,耗时"+(System.currentTimeMillis() - start));
}
}
2.我们在此基础上升级:
import java.util.Arrays;
import java.util.List;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.Executor;
import java.util.concurrent.Executors;
import java.util.stream.Collectors;
public class PriceDemo {
private List<Shop> 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("shop12"),
new Shop("shop13"),
new Shop("shop14"),
new Shop("shop15"),
new Shop("shop16"),
new Shop("shop17"),
new Shop("shop18"));
public List<String> findPrices(String product){
return shops.stream().parallel().map(shop->
String.format("%s price is %.2f", shop.getName(), shop.getPrice(product)))
.collect(Collectors.toList());
}
// public List<String> findPrice(String product){
// List<CompletableFuture<String>> priceFuture = shops.stream()
// .map(shop -> CompletableFuture
// .supplyAsync(()-> String.format("%s price is %.2f", shop.getName(), shop.getPrice(product))))
// .collect(Collectors.toList());
// return priceFuture.stream().map(CompletableFuture::join).collect(Collectors.toList());
// }
public List<String> findPriceExecutorsCompletableFuture(String product){
Executor executor = Executors.newFixedThreadPool(Math.min(shops.size(), 100));
List<CompletableFuture<String>> priceFuture = shops.stream()
.map(shop -> CompletableFuture
.supplyAsync(()-> String.format("%s price is %.2f", shop.getName(), shop.getPrice(product)), executor))
.collect(Collectors.toList());
return priceFuture.stream().map(CompletableFuture::join).collect(Collectors.toList());
}
public static void main(String[] args) {
PriceDemo priceDemo = new PriceDemo();
long start = System.currentTimeMillis();
System.out.println(priceDemo.findPrices("iphone 7"));
System.out.println("服务耗时:"+(System.currentTimeMillis()-start));
start = System.currentTimeMillis();
System.out.println(priceDemo.findPriceExecutorsCompletableFuture("iphone 7"));
System.out.println("服务耗时:"+(System.currentTimeMillis()-start));
}
}
我们发现threadPool结合CompletableFuture可以大大的提高性能。项目中哪些地方可以用到哪?
A simple C++11 Thread Pool implementation
最近提交(Master分支:2 个月前 )
9a42ec13 - 10 年前
fcc91415 - 10 年前
AtomGit 是由开放原子开源基金会联合 CSDN 等生态伙伴共同推出的新一代开源与人工智能协作平台。平台坚持“开放、中立、公益”的理念,把代码托管、模型共享、数据集托管、智能体开发体验和算力服务整合在一起,为开发者提供从开发、训练到部署的一站式体验。
更多推荐


所有评论(0)