0

Здравствуйте! Тут недавно вопрос задавал о сохранении данных из потока через интернет. Разобрался. Но на следующем этапе снова вопрос возник. Суть такая: То, что я хочу реализовать можно назвать консольной качалкой. Один из входных параметров - количество потоков в которых будут качаться отдельные файлы. Дак вот подозреваю, что нужно использовать ThreadPool, но он не работает корректно. То есть подразумевается, что цикл, который в классе Download создаст какое то количество потоков (в данном случае 5 - по количеству ссылок в dictionary), а часть из них поставит в очередь. Код внизу. В dictionary - ссылки на ресурсы и через пробел имя под которым ресурс будет сохранен на локальный компьютер (вид: http://example.com/ first.zip). Плюс ко всему, не уверен, что async await в полной мере использую (например стоит ли в данном случае асинхронными делать методы Read и Write класса Stream и без маркеров await все так же работает)

class Program
{
    static void Main(string[] args)
    {
        GetDownload gd = new GetDownload();
        Console.WriteLine("main {0}", Thread.CurrentThread.ManagedThreadId);

        gd.GetSource();

        ThreadPool.SetMaxThreads(4, 4);
        ThreadPool.QueueUserWorkItem(gd.DownLoad);

        Console.ReadKey();
    }
}
class GetDownload
{

    Dictionary<string, string> dictionary = new Dictionary<string, string>();
    Regex r = new Regex(@"((?<value_InputSource>[http]\S+)\s(?<key_OutFile>\S+))");
    public  void GetSource()
    {
        foreach (string line in File.ReadLines(@"C:\Sources.txt"))
        {              
               var regex = r.Matches(line);
                    if (regex.Count > 0)
                    {
                        string Source = regex[0].Groups["value_InputSource"].Value;
                        string OutFile = regex[0].Groups["key_OutFile"].Value;
                        dictionary.Add(OutFile, Source);
                    }   
        }           

    }

    public   void DownLoad(object obj)
    {

        HttpWebRequest request;
        HttpWebResponse response;
        for (int i = 0; i < dictionary.Values.Count; i++)
        {
         request=    HttpWebRequest.CreateHttp(dictionary.Values.ElementAt(i));
           response  = (HttpWebResponse)request.GetResponse();
                 Task.Factory.StartNew(async() => {

            using (Stream stream = response.GetResponseStream())
                {
                    using (FileStream fileSave = File.OpenWrite(dictionary.Keys.ElementAt(i)))
                    {
                        Stopwatch sw = new Stopwatch();
                        sw.Start();
                        byte[] buffer = new byte[2048];
                        int readNow = 0;
                        int totalDownloaded = 0;
                        while ((readNow =  stream.Read(buffer, 0, buffer.Length)) > 0)
                        {

                                  fileSave.Write(buffer, 0, readNow);
                            totalDownloaded += readNow;
                            var expectedTime = totalDownloaded / 1024.0 / 1;
                            var actualTime = (sw.Elapsed).TotalSeconds;
                            if (expectedTime > actualTime)
                              await  Task.Delay(TimeSpan.FromSeconds(expectedTime - actualTime));
                            Console.WriteLine("Уже закачано: {0}, Скорость: {1}, Поток: {2}", totalDownloaded, totalDownloaded / (actualTime * 1024), Thread.CurrentThread.ManagedThreadId);                         
                        }
                        sw.Stop();
                        Console.WriteLine("Загрузка завершена");
                    }
                }
            });
        }            
      }            
    }
3
  • Не понятно, что именно вы пытаетесь достичь. Вы хотите именно чтобы на загрузку каждого файла создавался отдельный поток? И хотите сделать это именно по средством ThreadPool? 5 дек 2016 в 19:56
  • @Yury Bakharev, Смотрите, например, в у меня есть 10 ссылок на какие-то файлы в интернете. Я хочу закачать эти 10 файлов так, чтобы у меня было всегда открыто не более 4 потоков. То есть сначала создается 4 потока и качаются первые 4 файла из списка. Каждый файл в одном потоке. Как только какой то файл загрузился например в 3 потоке, то есть, поток стал свободным - этот поток тут же начинает обрабатывать следующую по списку ссылку, т.е. пятую. И так далее, пока не загрузятся все. Средствами ThreadPool это реализовать или какими то другими - мне без разницы. Может быть есть способ удобнее 6 дек 2016 в 3:40
  • 3
    Всё настолько неправильно, что даже не знаю, с чего начать... 1) ThreadPool не предназначен для выполнения длительных блокирующих операций (таких как скачивание файлов); 2) По умолчанию, минимальное количество потоков в пуле равно количеству логических CPU, а изменение этого значения как в большую, так и в меньшую сторону снижает производительность; 3) Ваши потоки 99% времени ровным счетом ничего не делают: они находятся в блокировке, ожидая ответа либо от сетевого драйвера, либо от драйвера файловой системы; 4) Для выполнения асинхронных I/O операций дополнительные потоки вообще не нужны.
    – Raider
    9 дек 2016 в 23:07

2 ответа 2

-1

Проблема в том что вы синхронно вызываете "Синхронную" процедуру!

static Dictionary<string, string> dictionary = new Dictionary<string, string>();
    static void Main(string[] args)
    {
        foreach (var url in dictionary)
        {
            WaitCallback waite = new System.Threading.WaitCallback(DownLoad);
            WaitCallbacObject waiteCallbackObject = new WaitCallbacObject() { Path = url.Key };
            ThreadPool.SetMaxThreads(4, 4);
            ThreadPool.QueueUserWorkItem(waite, waiteCallbackObject);
        }
    }
    private static async void DownLoad(object state)
    {

        //Сюда в state вы передати url и сделаете вашу загрузку
        // Этот метод должен быть асинхронным
    }
2
  • ну если он и так в отдельном потоке, можно и без async
    – Grundy
    6 дек 2016 в 10:15
  • Окей. Я почитаю про waitcallback, разбирусь с этой информацией, если что-то будет не ясно - спрошу здесь. Спасибо за ответы 6 дек 2016 в 11:15
-1

Вдогонку к моему комментарию в оригинальном посте: вот решение, которое использует только асинхронные I/O операции, не создавая лишних потоков (т.е. без ThreadPool.QueueUserWorkItem и Task.Factory.StartNew):

public class Program : IProgress<DownloadProgress>
{
    public const int MaxSimultaneousDownloads = 3; // Маскимальное кол-во одновременных загрузок.
    public const int BufferSize = 512 * 1024; // 512 KB

    public static readonly IEnumerable<string> Links = new[]
    {
        "http://download.microsoft.com/download/b/a/4/ba4a7e71-2906-4b2d-a0e1-80cf16844f5f/dotnetfx45_full_x86_x64.exe",
        "http://download.microsoft.com/download/9/5/A/95A9616B-7A37-4AF6-BC36-D6EA96C8DAAE/dotNetFx40_Full_x86_x64.exe",
        "http://download.microsoft.com/download/6/0/f/60fc5854-3cb8-4892-b6db-bd4f42510f28/dotnetfx35.exe",
        "http://download.microsoft.com/download/a/3/f/a3f1bf98-18f3-4036-9b68-8e6de530ce0a/NetFx64.exe",
    };

    // Единственный блокирующий вызов - без него программа завершится раньше чем нужно.
    public static void Main(string[] args) => new Program().RunAsync().GetAwaiter().GetResult();

    public async Task RunAsync()
    {
        // Кол-во одновременных соединений ограничивается не ThreadPool, а ServicePointManager!
        ServicePointManager.DefaultConnectionLimit = MaxSimultaneousDownloads;

        var downloadTasks = new List<Task<string>>(MaxSimultaneousDownloads);
        var results = new List<string>();

        using (var httpClient = new HttpClient())
        using (var linksEnumerator = Links.GetEnumerator())
        {
            do
            {
                // Запускаем Task-и и добавляем их в список.
                bool anyLinksLeft = false;
                while (downloadTasks.Count < MaxSimultaneousDownloads && linksEnumerator.MoveNext())
                {
                    downloadTasks.Add(DownloadFileAsync(httpClient, linksEnumerator.Current, this));
                    anyLinksLeft = true;
                }

                Console.Title = $"Downloading: {downloadTasks.Count} file(s)";

                // Когда downloadTasks заполнен, ждем завершения любого Task-а из списка (хотя бы одного).
                if (downloadTasks.Count == MaxSimultaneousDownloads || !anyLinksLeft)
                {
                    var downloadTask = await Task.WhenAny(downloadTasks);
                    downloadTasks.Remove(downloadTask);
                    results.Add(await downloadTask);
                }

                // Прокручиваем цикл пока Task-и не закончатся.
            } while (downloadTasks.Any());
        }

        Console.WriteLine("Done!");
        results.ForEach(Console.WriteLine);
        Console.ReadKey();
    }

    public void Report(DownloadProgress progress)
    {
        Console.WriteLine(progress);
    }

    public static async Task<string> DownloadFileAsync(HttpClient httpClient, string link,
        IProgress<DownloadProgress> progress = null)
    {
        progress?.Report(new DownloadProgress(link, null, 0, null, 0));

        // Делаем асинхронный запрос на получение заголовков HTTP ответа (пока что только заголовков).
        var response = await httpClient.GetAsync(link, HttpCompletionOption.ResponseHeadersRead);

        // Content-Length может быть неизвестен!
        IEnumerable<string> contentLengths;
        var length = response.Headers.TryGetValues("Content-Length", out contentLengths)
            ? contentLengths.Select(long.Parse).First()
            : (long?) null;

        var stopwatch = Stopwatch.StartNew();

        using (var downloadStream = await response.Content.ReadAsStreamAsync())
        using (var fileStream = File.Create(Path.Combine(Path.GetTempPath(), Path.GetFileName(link))))
        {
            // В цикле читаем и пишем полученные байты асинхронно, пока stream не закончится.
            int read;
            var buffer = new byte[BufferSize];
            while ((read = await downloadStream.ReadAsync(buffer, 0, buffer.Length)) > 0)
            {
                stopwatch.Stop();

                await fileStream.WriteAsync(buffer, 0, read);

                // Репортим прогресс.
                progress?.Report(new DownloadProgress(link, fileStream.Name, fileStream.Length, length,
                    fileStream.Length / stopwatch.Elapsed.TotalSeconds));

                stopwatch.Start();
            }

            return fileStream.Name;
        }
    }
}

// Используем структуру чтобы избежать лишних аллокацией в куче.
public struct DownloadProgress
{
    private const double OneMegabyte = 1024 * 1024;

    public DownloadProgress(string link, string filePath,
        long downloaded, long? size, double bytesPerSecond)
    {
        Link = link;
        FilePath = filePath;
        Downloaded = downloaded;
        Size = size;
        BytesPerSecond = bytesPerSecond;
    }

    public string Link { get; }
    public string FilePath { get; }
    public long Downloaded { get; }
    public long? Size { get; }
    public double BytesPerSecond { get; }

    public override string ToString()
    {
        var sb = new StringBuilder();
        sb.AppendFormat("{0}: {1:F2} MB of ", Path.GetFileName(Link), Downloaded / OneMegabyte);
        if (Size != null)
        {
            sb.AppendFormat("{0:F2} MB downloaded ", Size / OneMegabyte);
        }
        else
        {
            sb.Append("? MB downloaded ");
        }
        if (Size == null || Downloaded < Size)
        {
            sb.AppendFormat("({0:F2} MB/s)", BytesPerSecond / OneMegabyte);
        }
        else
        {
            sb.Append("- Done!");
        }
        return sb.ToString();
    }
}
2
  • ого, вот это решение. Есть, куда стремиться, спасибо. А размер буффера из каких соображений выбран именно 512 кб? 10 дек 2016 в 16:06
  • С моим интернет соединением downloadStream.ReadAsync возвращает не больше 350 КБ за раз, так что я просто округлил в большую сторону. Я бы добавил еще один цикл, чтобы buffer записывался на диск только тогда, когда он полностью заполнится, но не хотелось усложнять код.
    – Raider
    10 дек 2016 в 18:24

Ваш ответ

By clicking “Отправить ответ”, you agree to our terms of service and acknowledge you have read our privacy policy.

Всё ещё ищете ответ? Посмотрите другие вопросы с метками или задайте свой вопрос.