ConcurrentStack的问题在 .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（并行 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)
{
    ...
}

条件立即为 false，因此您实际上从未等待线程。反过来，这意味着 errors 将包含迄今为止运行的任何线程的错误。

但是，您的代码还有其他问题，但手动创建线程可能是最大的问题。由于您使用的是 .NET 4.0，因此应该使用任务或 PLINQ 进行异步处理。使用 PLINQ，您的验证可以实现为：

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