Java虛擬機(jī)并發(fā)編程中運(yùn)算結(jié)果不確定問題

方兆國(guó)

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


Java虛擬機(jī)編程技術(shù)大家談---多核和并發(fā)編程
我在這個(gè)活動(dòng)中獲得了《Java虛擬機(jī)并發(fā)編程 》圖書1本.看了一下,照著書上的例子作了一下,發(fā)現(xiàn)了一個(gè)問題.

在書第18頁的那個(gè)關(guān)于線程池的例子,我在原例子上略作了一些改動(dòng),并沒有影響到核心問題,但是每次運(yùn)行的結(jié)果都是不確定的

以下是我的測(cè)試代碼

這個(gè)類是從服務(wù)器獲取數(shù)據(jù)文件,例子中的網(wǎng)址是雅虎財(cái)經(jīng)的,我現(xiàn)在打開那個(gè)連接已經(jīng)無效了,因此使用本地的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. }
    

復(fù)制代碼

這個(gè)類就是自己建立線程池來處理并發(fā)的請(qǐng)求,以前不是并發(fā)的時(shí)候完全沒有問題,改成并發(fā)之后,每次運(yùn)行的結(jié)果都不一樣

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. }
    

復(fù)制代碼

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. }

復(fù)制代碼

craaazy123

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

方兆國(guó)

craaazy123 發(fā)表于 2013-07-18 11:53
看看這樣行不行？  
for (final Future vaFuture : valFutures) {
      if(vaFuture.isDone()) {

額,謝謝你

gnah

這不是線程安全的問題,而是線程執(zhí)行順序不確定

zavakid

這樣的場(chǎng)景，推薦使用 CompletionService ，然后用 take 來獲得 future，這樣可以最快拿到結(jié)果。

方兆國(guó)

craaazy123 發(fā)表于 2013-07-18 11:53
看看這樣行不行？  
for (final Future vaFuture : valFutures) {
      if(vaFuture.isDone()) {

我試了不可以

但是在另外一個(gè)例子中同樣的方法是成功的.感覺這個(gè)問題是由于IO速度慢造成的,不過我是本機(jī)測(cè)試都這樣,如果換成是直接從網(wǎng)上獲取數(shù)據(jù)那不更慢

方兆國(guó)

gnah 發(fā)表于 2013-07-18 12:58
這不是線程安全的問題,而是線程執(zhí)行順序不確定

最后的數(shù)據(jù)結(jié)果不一致,應(yīng)該是由數(shù)據(jù)丟失了,我做了一個(gè)沒有磁盤IO的例子，不存在這個(gè)問題
我試一下用虛擬磁盤吧

方兆國(guó)

這是用虛擬盤軟件制作了一個(gè)虛擬磁盤，盤符是F：


將所有的文件都放在虛擬盤中，運(yùn)算結(jié)果還是不一致

方兆國(guó)

試一下樓上建議的CompletionService吧
感覺這個(gè)問題很怪異，我試了另外一個(gè)例子完全沒有問題，是個(gè)算素?cái)?shù)的，沒有磁盤IO操作，是運(yùn)算密集型的例子

方兆國(guó)

回復(fù) 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.         }

復(fù)制代碼

我從網(wǎng)上看了一個(gè)例子，是這么個(gè)用法，可是運(yùn)行起來還是老毛病，結(jié)果總是老變