.net c#4 中的 ConcurrentStack<T> 问题

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

这是 Task Parallel Library 和 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);
}