Parallel.ForEach CancellationTokenSource 未停止答案

【问题标题】：Parallel.ForEach CancellationTokenSource not StoppingParallel.ForEach CancellationTokenSource 未停止
【发布时间】：2014-06-15 20:43:15
【问题描述】：

我目前正在编写一个 ProxyChecker 库。我正在使用一个有趣的 Parallel.ForEach 循环来检查所有代理的线程。我正在使用CancellationTokenSource（cts）进行软中止（使用cts.Cancel()）。正如您在下面的代码中看到的那样，我添加了一些“测试代码”，它将当前线程写入控制台。

这是你需要的代码：

private void CheckProxies(string[] proxies, int timeout, int threads, string domainToCheckWith)
        {
            _cts = new CancellationTokenSource();
            int checkedProxyCount = 0, uncheckedProxyCount = proxies.Length, goodProxies = 0, badProxies = 0;
            mainThread = new Thread(() =>
            {
                try
                {
                    Parallel.ForEach(proxies, new ParallelOptions {MaxDegreeOfParallelism = threads, CancellationToken = _cts.Token}, prox =>
                    {
                        Interlocked.Increment(ref running);
                        Console.WriteLine("thread running: {0}", running);
                        try
                        {
                            _cts.Token.ThrowIfCancellationRequested();
                            if (CheckProxy(prox, domainToCheckWith, timeout))
                            {
                                Interlocked.Increment(ref checkedProxyCount);
                                Interlocked.Increment(ref goodProxies);
                                Interlocked.Decrement(ref uncheckedProxyCount);
                            }
                            else
                            {
                                Interlocked.Increment(ref checkedProxyCount);
                                Interlocked.Decrement(ref uncheckedProxyCount);
                                Interlocked.Increment(ref badProxies);
                            }
                            _cts.Token.ThrowIfCancellationRequested();
                            OnUpdate(uncheckedProxyCount, checkedProxyCount, goodProxies, badProxies);
                        }
                        catch (OperationCanceledException ex) {}
                        catch (ObjectDisposedException ex) {}
                        catch (Exception ex)
                        {
                            OnLog(ex.Message, Color.Red);
                        }
                        finally
                        {
                            Console.WriteLine("thread running: {0}", running);
                            Interlocked.Decrement(ref running);
                        }
                    });
                }
                catch (OperationCanceledException ex) {}
                catch (ObjectDisposedException ex) {}
                catch (Exception ex)
                {
                    OnLog(ex.Message, Color.Red);
                }
                finally
                {
                    isRunning = false;
                    OnComplete();
                }
            });
            mainThread.Start();
        }

输出（我删了几行，因为给你完整的代码没用）

thread running: 1
thread running: 1
thread running: 2
thread running: 2

//Slowly going up to  50

thread running: 50
thread running: 50
thread running: 50

//Staying at 50 till I press stop

thread running: 50
thread running: 50
thread running: 50
thread running: 50
thread running: 50
thread running: 49
thread running: 48
thread running: 47
thread running: 46

//Going down...

thread running: 17
thread running: 16
thread running: 15
thread running: 14
thread running: 13
thread running: 12
thread running: 11
thread running: 10
thread running: 10
thread running: 8
thread running: 7
thread running: 6
thread running: 5
thread running: 4

然后它在 4 或 3 或 2 处停止（每次都不同）。我等了几分钟，但在 Parallel.ForEach 执行后它并没有关闭，也没有代码。

请求的超时时间为 5000，线程数为 50。

这是检查的其他代码：

private bool CheckProxy(string proxy, string domainToCheckWith, int timeout)
{
    try
    {
        WebRequest req = WebRequest.Create(domainToCheckWith);
        req.Proxy = new WebProxy(proxy);
        req.Timeout = timeout;
        var response = (HttpWebResponse) req.GetResponse();
        string responseString = ReadResponseString(response);

        if (responseString.Contains("SOMETHING HERE"))
        {
            OnGoodProxy(proxy);
            return true;
        }
        if (responseString.Contains("SOMEOTHERTHINGHERE"))
        {
            OnBadProxy(proxy);
            return false;
        }
        OnBadProxy(proxy);
        return false;
    }
    catch (WebException ex)
    {
        OnBadProxy(proxy);
        return false;
    }
    catch (Exception ex)
    {
        OnLog(ex.Message, Color.Red);
        return false;
    }
}

停止功能：

public void StopChecking()
{
    try
    {
        if (_cts != null && mainThread.IsAlive)
        {
            if (_cts.IsCancellationRequested)
            {
                mainThread.Abort();
                OnLog("Hard aborting Filter Threads...", Color.DarkGreen);
                while (mainThread.IsAlive) ;
                OnComplete();
                isRunning = false;
            }
            else
            {
                _cts.Cancel();
                OnLog("Soft aborting Filter Threads...", Color.DarkGreen);
            }
        }
    }
    catch (Exception ex)
    {
        OnLog(ex.Message, Color.Red);
    }
}

重要编辑：

我在 CeckProxy 函数中添加了这个：

        Stopwatch sw = new Stopwatch();
        sw.Start();
        string responseString = new StreamReader(response.GetResponseStream()).ReadToEnd();
        sw.Stop();

这是最后几个线程的结果：

thread running: 6
4449
thread running: 5
72534
thread running: 4
180094
thread running: 3

为什么这么长？我的意思是 180 秒？！

【问题讨论】：

你是否真的调用过_cts.Cancel()。我在任何地方都看不到它。
@brumScouse 我没有添加那部分，但我可以。我会在一秒钟内编辑 OP
我会冒险猜测 StopChecking 中的 _cts 必须限定为该方法所在的类？我注意到每次在您的 CheckProxies 方法中都会更新 _cts，这是故意的吗？我的意思是这里的 _cts 必须与 StopChecking 中的相同（我只是在检查）
private CancellationTokenSource _cts; 这是类中的一个字段！是的，这是故意的。每次我想处理新代理时，我都必须生成一个新的 CancellationTokenSource
我想知道如何将线程更改为任务并通过取消令牌源。我只是想知道正在更新的线程是否捕获了 cts...

标签： c# parallel-processing cancellationtokensource

【解决方案1】：

你可以尝试 lock inside 尝试锁定对象

Object lockObject = new Object();
try
{
    Parallel.ForEach(proxies, new ParallelOptions {MaxDegreeOfParallelism = threads, CancellationToken = _cts.Token}, prox =>
    {
        Interlocked.Increment(ref running);
        Console.WriteLine("thread running: {0}", running);
        try
        {
            lock(lockObject)
            {
                //code.............
            }
        }
        catch
        {
        }
    }
}
catch
{
}

【讨论】：

它有效。但现在它不再是真正的多线程，因为它一直等到前一个线程完成/锁被释放！所以这实际上对我没有帮助！
你必须做这样的事情 ParallelOptions pOptions = new ParallelOptions();比在你的 try catch 块内或者在 if else 内你必须调用 pOptions.CancellationToken.ThrowIfCancellationRequested();或者你也可以有 ...,proxy, (i, loopstate)=> 而不是在 try catch 你需要 loopstate.Stop();
@Davidson 我也试过了，但结果是一样的

【解决方案2】：

好的，我自己想通了。

我现在连续读取响应并使用秒表（和 request.ReadWriteTimeout）检查读取部分是否在达到特定时间（在我的情况下为 readTimeout）后停止。代码

HttpWebRequest req = (HttpWebRequest)WebRequest.Create(domainToCheckWith);
            req.Proxy = new WebProxy(proxy);
            req.Timeout = timeout;
            req.ReadWriteTimeout = readTimeout;
            req.Headers.Add(HttpRequestHeader.AcceptEncoding, "deflate,gzip");
            req.AutomaticDecompression = DecompressionMethods.Deflate | DecompressionMethods.GZip;

            byte[] responseByte = new byte[1024];
            string responseString = string.Empty;

            sw.Start();
            using (WebResponse res = req.GetResponse())
            {
                using (Stream stream = res.GetResponseStream())
                {
                    while (stream.Read(responseByte, 0, responseByte.Length) > 0)
                    {
                        responseString += Encoding.UTF8.GetString(responseByte);
                        if(sw.ElapsedMilliseconds > (long)timeout)
                            throw new WebException();
                    }

                }
            }
            sw.Stop();

【讨论】：