Проблема с ConcurrentStack <T>в .net c # 4

Question

Этот класс предназначен для получения списка URL-адресов, их сканирования, а затем возврата списка тех, которые не работают.Это использует многократные потоки, чтобы избежать навсегда занять длинные списки.

Моя проблема в том, что, даже если я заменил фактическое сканирование URL-адресов на тестовую функцию, которая возвращает ошибки всех URL-адресов, класс возвращает переменное количество ошибок.Проблема заключается либо в ConcurrentStack.TryPop () или .Push (), но я не могу на всю жизнь понять, почему.Они предположительно безопасны для многопоточной обработки, и я тоже пытался их заблокировать, там никакой помощи.

Кто-нибудь может мне объяснить, что я делаю неправильно?У меня нет большого опыта работы с несколькими темами ..

public class UrlValidator
{
    private const int MAX_THREADS = 10;
    private List<Thread> threads = new List<Thread>();
    private ConcurrentStack<string> errors = new ConcurrentStack<string>();
    private ConcurrentStack<string> queue = new ConcurrentStack<string>();

    public UrlValidator(List<string> urls)
    {
        queue.PushRange(urls.ToArray<string>());
    }

    public List<string> Start()
    {
        threads = new List<Thread>();

        while (threads.Count < MAX_THREADS && queue.Count > 0)
        {
            var t = new Thread(new ThreadStart(UrlWorker));
            threads.Add(t);
            t.Start();
        }

        while (queue.Count > 0) Thread.Sleep(1000);

        int runningThreads = 0;
        while (runningThreads > 0)
        {
            runningThreads = 0;
            foreach (Thread t in threads) if (t.ThreadState == ThreadState.Running) runningThreads++;
            Thread.Sleep(100);
        }

        return errors.ToList<string>();
    }

    private void UrlWorker()
    {
        while (queue.Count > 0)
        {
            try 
            {
                string url = "";
                if (!queue.TryPop(out url)) continue;
                if (TestFunc(url) != 200) errors.Push(url);
            }
            catch
            {
                break;
            }
        }
    }

    private int TestFunc(string url)
    {
        Thread.Sleep(new Random().Next(100));
        return -1;
    }
}

Answer 1

Это то, что было бы неплохо для параллельной библиотеки задач и PLINQ (Parallel LINQ). Посмотрите пример того, насколько проще будет, если вы позволите .NET делать свое дело:

public IEnumerable<string> ProcessURLs(IEnumerable<string> URLs)
{
    return URLs.AsParallel()
        .WithDegreeOfParallelism(10)
        .Where(url => testURL(url));
}

private bool testURL(string URL)
{
    // some logic to determine true/false
    return false;
}

По возможности, вы должны разрешать библиотекам .NET обеспечивать любое необходимое управление потоками. В целом TPL отлично подходит для этого, но поскольку вы просто трансформируете одну коллекцию элементов, PLINQ хорошо подходит для этого. Вы можете изменить степень параллелизма (я бы порекомендовал установить его меньше, чем максимальное количество одновременных TCP-соединений), и вы можете добавить несколько условий, как позволяет LINQ. Автоматически запускается параллельно и не требует управления потоками.

Answer 2

Ваша проблема не имеет ничего общего с ConcurrentStack, а скорее с циклом, в котором вы проверяете наличие потоков:

int runningThreads = 0;
while (runningThreads > 0)
{
    ...
}

Условие сразу ложно, поэтому вы никогда не будете ждать потоков. В свою очередь, это означает, что errors будет содержать ошибки от тех потоков, которые были запущены до сих пор.

Тем не менее, у вашего кода есть другие проблемы, но создание потоков вручную, вероятно, является самым большим. Поскольку вы используете .NET 4.0, вы должны использовать задачи или PLINQ для асинхронной обработки. Используя PLINQ, ваша проверка может быть реализована как:

public IEnumerable<string> Validate(IEnumerable<string> urls)
{ 
    return urls.AsParallel().Where(url => TestFunc(url) != 200);
}