Первый раз, когда создавался метод, который использует параллельное linq и исключение из памяти

Question

Я написал метод для загрузки данных из Интернета и сохранения его в моей базе данных. Я написал это с помощью PLINQ, чтобы воспользоваться моим многоядерным процессором и потому, что он загружает тысячи разных файлов за очень короткий промежуток времени. Я добавил комментарии ниже в свой код, чтобы показать, где он останавливается, но программа просто сидит там, и через некоторое время я получаю исключение из памяти. Это мой первый раз, когда я использовал TPL и PLINQ, я очень смущен, поэтому я действительно могу использовать некоторые советы о том, что делать, чтобы исправить это.

ОБНОВЛЕНИЕ: Я узнал, что постоянно получаю веб-представление, потому что веб-клиенту не хватает времени. Я исправил это, увеличив максимальное количество соединений в соответствии с 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

Async лямбда работать как асинхронные методы в одном отношении: они не заканчивают синхронно, но они возвращают задачу. В своем параллельном цикле вы просто выполняете задачи так быстро, как можете. Эти задачи хранятся в памяти и других ресурсах, таких как соединения с БД.

Простейшим решением является, вероятно, просто использовать синхронные транзакции базы данных. Это не приведет к потере пропускной способности, поскольку база данных не может иметь дело с большим количеством одновременных DML в любом случае.