第一次創建一個使用並行linq和內存異常的方法

Question

我寫了一個方法從互聯網下載數據並將其保存到我的數據庫。我使用PLINQ編寫了這個文件，以充分利用我的多核處理器，並且因爲它可以在很短的時間內下載數千個不同的文件。我在代碼中添加了以下注釋，以顯示它停止的位置，但程序剛好位於此處，過了一段時間後，出現內存不足異常。這是我第一次使用TPL和PLINQ，我非常困惑，所以我可以真正使用一些建議來解決這個問題。

更新：我發現我經常收到引發WebException因爲WebClient的是超時。我通過根據this answer here增加最大連接數來解決這個問題。然後我發現連接沒有打開，我通過使用this answer here來修復它。即使它是本地sql服務器，我現在也會爲數據庫獲取連接超時錯誤。我還沒有能夠得到我的任何代碼運行，所以我可以完全使用一些建議

static void Main(string[] args) 
    { 
     try 
     { 
      while (true) 
      { 
       // start the download process for market info 
       startDownload(); 
      } 
     } 
     catch (Exception ex) 
     { 
      Console.WriteLine(ex.Message); 
      Console.WriteLine(ex.StackTrace); 
     } 
    } 

    public static void startDownload() 
    { 
     DateTime currentDay = DateTime.Now; 
     List<Task> taskList = new List<Task>(); 

     if (Helper.holidays.Contains(currentDay) == false) 
     { 
      List<string> markets = new List<string>() { "amex", "nasdaq", "nyse", "global" }; 

      Parallel.ForEach(markets, market => 
      { 
       Downloads.startInitialMarketSymbolsDownload(market); 
      } 
      ); 

      Console.WriteLine("All downloads finished!"); 
     } 

     // wait 24 hours before you do this again 
     Task.Delay(TimeSpan.FromHours(24)).Wait(); 
    } 

public static void startInitialMarketSymbolsDownload(string market) 
    { 
     try 
     { 
      List<string> symbolList = new List<string>(); 
      symbolList = Helper.getStockSymbols(market); 

      var historicalGroups = symbolList.AsParallel().Select((x, i) => new { x, i }) 
         .GroupBy(x => x.i/100) 
         .Select(g => g.Select(x => x.x).ToArray()); 

      historicalGroups.AsParallel().ForAll(g => getHistoricalStockData(g, market)); 
     } 
     catch (Exception ex) 
     { 
      Console.WriteLine(ex.Message); 
      Console.WriteLine(ex.StackTrace); 
     } 
    } 

public static void getHistoricalStockData(string[] symbols, string market) 
    { 
     // download data for list of symbols and then upload to db tables 
     Uri uri; 
     string url, line; 
     decimal open = 0, high = 0, low = 0, close = 0, adjClose = 0; 
     DateTime date; 
     Int64 volume = 0; 
     string[] lineArray; 
     List<string> symbolError = new List<string>(); 
     Dictionary<string, string> badNameError = new Dictionary<string, string>(); 

     Parallel.ForEach(symbols, symbol => 
       { 
         url = "http://ichart.finance.yahoo.com/table.csv?s=" + symbol + "&a=00&b=1&c=1900&d=" + (DateTime.Now.Month - 1) + "&e=" + DateTime.Now.Day + "&f=" + DateTime.Now.Year + "&g=d&ignore=.csv"; 
         uri = new Uri(url); 

         using (dbEntities entity = new dbEntities()) 
         using (WebClient client = new WebClient()) 
         using (Stream stream = client.OpenRead(uri)) 
         using (StreamReader reader = new StreamReader(stream)) 
         { 
          while (reader.EndOfStream == false) 
          { 
           line = reader.ReadLine(); 
           lineArray = line.Split(','); 

           // if it isn't the very first line 
           if (lineArray[0] != "Date") 
           { 
            // set the data for each array here 
            date = Helper.parseDateTime(lineArray[0]); 
            open = Helper.parseDecimal(lineArray[1]); 
            high = Helper.parseDecimal(lineArray[2]); 
            low = Helper.parseDecimal(lineArray[3]); 
            close = Helper.parseDecimal(lineArray[4]); 
            volume = Helper.parseInt(lineArray[5]); 
            adjClose = Helper.parseDecimal(lineArray[6]); 

            switch (market) 
            { 
             case "nasdaq": 
              DailyNasdaqData nasdaqData = new DailyNasdaqData(); 
              var nasdaqQuery = from r in entity.DailyNasdaqDatas.AsParallel().AsEnumerable() 
                   where r.Date == date 
                   select new StockData { Close = r.AdjustedClose }; 

              List<StockData> nasdaqResult = nasdaqQuery.AsParallel().ToList(); // hits this line 
              break; 
             default: 
              break; 
            } 
           } 
          } 

          // now save everything 
          entity.SaveChanges(); 
         } 
       } 
     ); 
    } 

Answer 1

異步lambda表達式的工作就像在一個方面異步方法：他們不同步完成，但他們返回的任務。在你的並行循環中，你只需要儘可能快地生成任務。這些任務持有內存和其他資源，如數據庫連接。

最簡單的解決方法可能是使用同步數據庫提交。這不會導致吞吐量的損失，因爲數據庫無論如何都無法處理大量的併發DML。