Производитель / потребитель веб-сканера, использующий очередь с неизвестным размером

Question

Мне нужно сканировать родительские веб-страницы и их дочерние веб-страницы, и я следовал концепции производителя / потребителя с http://www.albahari.com/threading/part4.aspx#%5FWait%5Fand%5FPulse. Кроме того, я использовал 5 потоков, которые ставят в очередь и удаляют ссылки.

Любые рекомендациикак мне завершить / объединить все потоки после того, как все они закончили обработку очереди, учитывая, что длина очереди неизвестна?

Ниже приведена идея, как я ее кодировал.

static void Main(string[] args)
{
    //enqueue parent links here
    ...
    //then start crawling via threading
    ...
}

public void Crawl()
{
   //dequeue
   //get child links
   //enqueue child links
}

Answer 1

Если все ваши потоки бездействуют (т. Е. Ожидают в очереди), а очередь пуста, значит, все готово.

Простой способ справиться с тем, чтобы потоки использовали тайм-аут, когдаони пытаются получить доступ к очереди.Что-то вроде BlockingCollection.TryTake .Всякий раз, когда TryTake истекает, поток обновляет поле, чтобы сказать, как долго он простаивает:

while (!queue.TryTake(out item, 5000, token))
{
    if (token.IsCancellationRequested)
        break;
    // here, update idle counter
}

Затем вы можете иметь таймер, который запускается каждые 15 секунд или около того, чтобы проверять все бездействующие потоки.счетчики.Если все потоки простаивали в течение некоторого периода времени (возможно, минуты), таймер может установить токен отмены.Это убьет все темы.Ваша основная программа также может отслеживать токен отмены.

Кстати, вы можете сделать это без BlockingCollection и отмены.Вам просто нужно создать собственный механизм сигнализации отмены, и если вы используете блокировку в очереди, вы можете заменить синтаксис блокировки на Monitor.TryEnter и т. Д.

Есть несколько других способовсправиться с этим, хотя они потребуют серьезной реструктуризации вашей программы.

Answer 2

В конце вы можете поставить фиктивный токен и заставить нити выйти, когда они встретят этот токен. Как:

public void Crawl()
{
   int report = 0;
   while(true)
   {
       if(!(queue.Count == 0))      
       {   
          if(report > 0) Interlocked.Decrement(ref report);
          //dequeue     
          if(token == "TERMINATION")
             return;
          else
             //enqueue child links
       }
       else
       {              
          if(report == num_threads) // all threads have signaled empty queue
             queue.Enqueue("TERMINATION");
          else
             Interlocked.Increment(ref report); // this thread has found the queue empty
       }
    }
}

Конечно, я исключил блокировки для enqueue/dequeue операций.

Answer 3

Нет необходимости обрабатывать вещи производителя-потребителя вручную, если вы хотите использовать Task Parallel Library .При создании задач с параметром AttachToParent дочерние задачи будут связываться с родительской задачей таким образом, что она не будет завершена до тех пор, пока дочерние задачи не будут завершены.

class Program
{
    static void Main(string[] args)
    {
        var task = CrawlAsync("http://stackoverflow.com");
        task.Wait();
    }

    static Task CrawlAsync(string url)
    {
        return Task.Factory.StartNew(
            () =>
            {
                string[] children = ExtractChildren(url);
                foreach (string child in children)
                {
                    CrawlAsync(child);
                }
                ProcessUrl(url);
            }, TaskCreationOptions.AttachedToParent);
    }

    static string[] ExtractChildren(string root)
    {
      // Return all child urls here.
    }

    static void ProcessUrl(string url)
    {
      // Process the url here.
    }
}

Вы можете удалить некоторые явные задачилогика создания задачи с использованием Parallel.ForEach.

Answer 4

Потоки могут сигнализировать о том, что завершили свою работу, например, вызвать событие или вызвать делегата.

static void Main(string[] args)
{
//enqueue parent links here
...
//then start crawling via threading
...
}

public void X()
{
    //block the threads until all of them are here
}

public void Crawl(Action x)
{
    //dequeue
    //get child links
    //enqueue child links
    //call x()
}