Java虚拟机并发编程中运算结果不确定问题

方兆国

@duanjigang @send_linux
@hbsycw @redcap0 @zavakid @gnah


Java虚拟机编程技术大家谈---多核和并发编程
我在这个活动中获得了《Java虚拟机并发编程 》图书1本.看了一下,照着书上的例子作了一下,发现了一个问题.

在书第18页的那个关于线程池的例子,我在原例子上略作了一些改动,并没有影响到核心问题,但是每次运行的结果都是不确定的

以下是我的测试代码

这个类是从服务器获取数据文件,例子中的网址是雅虎财经的,我现在打开那个连接已经无效了,因此使用本地的Nginx提供文件

1. package com.fangzhaoguo.finance;
   
2. 
   
3. import java.io.BufferedReader;
   
4. import java.io.IOException;
   
5. import java.io.InputStreamReader;
   
6. import java.net.MalformedURLException;
   
7. import java.net.URL;
   
8. 
   
9. public class YahooFinance {
   
10. 
    
11.         final static int ID = 0;
    
12.         final static int Name = 1;
    
13. 
    
14.         private URL url;
    
15.         private BufferedReader reader;
    
16. 
    
17.         public YahooFinance(final String host) {
    
18.                 try {
    
19.                         url = new URL(host);
    
20.                         reader = new BufferedReader(new InputStreamReader(url.openStream()));
    
21.                 } catch (MalformedURLException e) {
    
22.                         e.printStackTrace();
    
23.                 } catch (IOException e) {
    
24.                         e.printStackTrace();
    
25.                 }
    
26.         }
    
27. 
    
28.         public double getPrice(final String ticker, int way) {
    
29.                 String data = null;
    
30.                 String[] dataItem;
    
31.                 double lastPrice = 0;
    
32. 
    
33.                 try {
    
34.                         while (null != (data = reader.readLine())) {
    
35.                                 dataItem = data.split(",");
    
36.                                 if (ticker.equals(dataItem[way])) {
    
37.                                         lastPrice = Double.valueOf(dataItem[dataItem.length - 1]);
    
38.                                         break;
    
39.                                 }
    
40.                         }
    
41.                 } catch (IOException e) {
    
42.                         e.printStackTrace();
    
43.                 }
    
44.                 return lastPrice;
    
45.         }
    
46. 
    
47.         public void close() {
    
48.                 try {
    
49.                         reader.close();
    
50.                 } catch (IOException e) {
    
51.                         e.printStackTrace();
    
52.                 }
    
53.         }
    
54. }
    

复制代码

这个类就是自己建立线程池来处理并发的请求,以前不是并发的时候完全没有问题,改成并发之后,每次运行的结果都不一样

1. package com.fangzhaoguo.finance;
   
2. 
   
3. import java.io.BufferedReader;
   
4. import java.io.FileNotFoundException;
   
5. import java.io.FileReader;
   
6. import java.io.IOException;
   
7. import java.util.ArrayList;
   
8. import java.util.HashMap;
   
9. import java.util.List;
   
10. import java.util.Map;
    
11. import java.util.concurrent.Callable;
    
12. import java.util.concurrent.ExecutionException;
    
13. import java.util.concurrent.ExecutorService;
    
14. import java.util.concurrent.Executors;
    
15. import java.util.concurrent.Future;
    
16. import java.util.concurrent.TimeUnit;
    
17. 
    
18. public class YahooFinanceNAV {
    
19. 
    
20.         String host, filename;
    
21.         YahooFinance finance;
    
22.         Map<String, Integer> stocks;
    
23. 
    
24.         public YahooFinanceNAV(String host, String filename) {
    
25.                 this.host = host;
    
26.                 this.filename = filename;
    
27.                 finance = new YahooFinance(host);
    
28.                 stocks = new HashMap<String, Integer>();
    
29.         }
    
30. 
    
31.         public void readTickers() {
    
32.                 try {
    
33.                         final BufferedReader reader = new BufferedReader(new FileReader(
    
34.                                         filename));
    
35.                         String info = null;
    
36.                         String[] infoItem;
    
37.                         while (null != (info = reader.readLine())) {
    
38.                                 infoItem = info.split(",");
    
39.                                 stocks.put(infoItem[0], Integer.valueOf(infoItem[1]));
    
40.                         }
    
41.                         reader.close();
    
42.                 } catch (FileNotFoundException e) {
    
43.                         e.printStackTrace();
    
44.                 } catch (IOException e) {
    
45.                         e.printStackTrace();
    
46.                 }
    
47.         }
    
48. 
    
49.         private int getPoolSize() {
    
50.                 return 4 + stocks.size() / Runtime.getRuntime().availableProcessors();
    
51.         }
    
52. 
    
53.         public double computeNAV() {
    
54.                 double NAV = 0.0;
    
55.                 int poolSize = getPoolSize();
    
56.                 List<Callable<Double>> partitions = new ArrayList<Callable<Double>>();
    
57.                 for (final String ticker : stocks.keySet()) {
    
58.                         partitions.add(new Callable<Double>() {
    
59.                                 public Double call() throws Exception {
    
60.                                         return stocks.get(ticker)
    
61.                                                         * finance.getPrice(ticker, YahooFinance.ID);
    
62.                                 }
    
63.                         });
    
64.                 }
    
65. 
    
66.                 ExecutorService executorService = Executors
    
67.                                 .newFixedThreadPool(poolSize);
    
68.                 try {
    
69.                         List<Future<Double>> valFutures = executorService.invokeAll(
    
70.                                         partitions, 10000, TimeUnit.SECONDS);
    
71.                         for (final Future<Double> vaFuture : valFutures) {
    
72.                                 NAV += vaFuture.get();
    
73.                         }
    
74.                 } catch (InterruptedException e) {
    
75.                         e.printStackTrace();
    
76.                 } catch (ExecutionException e) {
    
77.                         e.printStackTrace();
    
78.                 }
    
79.                 executorService.shutdown();
    
80.                 finance.close();
    
81.                 return NAV;
    
82.         }
    
83. }
    

复制代码

1. package com.fangzhaoguo.finance;
   
2. 
   
3. public class Main {
   
4. 
   
5.         public static void main(String[] args) {
   
6.                 YahooFinanceNAV yahooFinanceNAV = new YahooFinanceNAV(
   
7.                                 "http://localhost/table.csv", "stocks.dat");
   
8.                 yahooFinanceNAV.readTickers();
   
9.                 System.out.println(yahooFinanceNAV.computeNAV());
   
10.         }
    
11. }

复制代码

craaazy123

看看这样行不行？  
for (final Future<Double> vaFuture : valFutures) {
      if(vaFuture.isDone()) {
            NAV += vaFuture.get();
       }
}

方兆国

craaazy123 发表于 2013-07-18 11:53
看看这样行不行？  
for (final Future vaFuture : valFutures) {
      if(vaFuture.isDone()) {

额,谢谢你

gnah

这不是线程安全的问题,而是线程执行顺序不确定

zavakid

这样的场景，推荐使用 CompletionService ，然后用 take 来获得 future，这样可以最快拿到结果。

方兆国

craaazy123 发表于 2013-07-18 11:53
看看这样行不行？  
for (final Future vaFuture : valFutures) {
      if(vaFuture.isDone()) {

我试了不可以

但是在另外一个例子中同样的方法是成功的.感觉这个问题是由于IO速度慢造成的,不过我是本机测试都这样,如果换成是直接从网上获取数据那不更慢

方兆国

gnah 发表于 2013-07-18 12:58
这不是线程安全的问题,而是线程执行顺序不确定

最后的数据结果不一致,应该是由数据丢失了,我做了一个没有磁盘IO的例子，不存在这个问题
我试一下用虚拟磁盘吧

方兆国

这是用虚拟盘软件制作了一个虚拟磁盘，盘符是F：


将所有的文件都放在虚拟盘中，运算结果还是不一致

方兆国

试一下楼上建议的CompletionService吧
感觉这个问题很怪异，我试了另外一个例子完全没有问题，是个算素数的，没有磁盘IO操作，是运算密集型的例子

方兆国

回复 5# zavakid

1. public double computeNAV() {
   
2.                 double NAV = 0.0;
   
3.                 int poolSize = getPoolSize();
   
4.                 ExecutorService executorService = Executors
   
5.                                 .newFixedThreadPool(poolSize);
   
6.                 CompletionService<Double> completionService = new ExecutorCompletionService<Double>(
   
7.                                 executorService);
   
8.                 for (final String ticker : stocks.keySet()) {
   
9.                         completionService.submit(new Callable<Double>() {
   
10.                                 public Double call() throws Exception {
    
11.                                         return stocks.get(ticker)
    
12.                                                         * finance.getPrice(ticker, YahooFinance.ID);
    
13.                                 }
    
14.                         });
    
15.                 }
    
16.                 for (int i = 0; i < stocks.size(); i++) {
    
17.                         try {
    
18.                                 NAV += completionService.take().get();
    
19.                         } catch (InterruptedException | ExecutionException e) {
    
20.                                 e.printStackTrace();
    
21.                         }
    
22.                 }
    
23.                 executorService.shutdown();
    
24.                 finance.close();
    
25.                 return NAV;
    
26.         }

复制代码

我从网上看了一个例子，是这么个用法，可是运行起来还是老毛病，结果总是老变