知识库--Concurrency+ThreadPool+Executors(79)

public abstract class AbstractNAV {
public static Map<String, Integer> readTickers() throws IOException {
final BufferedReader reader =
new BufferedReader(new FileReader("C:\\Users\\ang\\Desktop\\stocks.txt"));
final Map<String, Integer> stocks = new HashMap<String, Integer>();
String stockInfo = null;
while((stockInfo = reader.readLine()) != null) {
final String[] stockInfoData = stockInfo.split(",");
final String stockTicker = stockInfoData[0];
final Integer quantity = Integer.valueOf(stockInfoData[1]);
stocks.put(stockTicker, quantity);
}
return stocks;
}
public void timeAndComputeValue()
throws ExecutionException, InterruptedException, IOException {
final long start = System.nanoTime();
final Map<String, Integer> stocks = readTickers();
final double nav = computeNetAssetValue(stocks);
final long end = System.nanoTime();
final String value = new DecimalFormat("$##,##0.00").format(nav);
System.out.println("Your net asset value is " + value);
System.out.println("Time (seconds) taken " + (end - start)/1.0e9);
}
public abstract double computeNetAssetValue(
final Map<String, Integer> stocks)
throws ExecutionException, InterruptedException, IOException;
}

public class Concurrent extends AbstractNAV {
public double computeNetAssetValue(final Map<String, Integer> stocks)
throws InterruptedException, ExecutionException {
final int numberOfCores = Runtime.getRuntime().availableProcessors();
final double blockingCoefficient = 0.9;
final int poolSize = (int) (numberOfCores / (1 - blockingCoefficient));
System.out.println("Number of Cores available is " + numberOfCores);
System.out.println("Pool size is " + poolSize);
final List<Callable<Double>> partitions = new ArrayList<Callable<Double>>();
for (final String ticker : stocks.keySet()) {
partitions.add(new Callable<Double>() {
public Double call() throws Exception {
return stocks.get(ticker) * YahooFinance.getPrice(ticker);
}
});
}

final ExecutorService executorPool = Executors.newFixedThreadPool(poolSize);
final List<Future<Double>> valueOfStocks =
executorPool.invokeAll(partitions, 10000, TimeUnit.SECONDS);
double netAssetValue = 0.0;
for (final Future<Double> valueOfAStock : valueOfStocks)
netAssetValue += valueOfAStock.get();

executorPool.shutdown();
return netAssetValue;
}

public static void main(final String[] args) throws ExecutionException, InterruptedException, IOException {
new Concurrent().timeAndComputeValue();
}
}