关于队列和线程安全的查询

Question

线程安全并不是我所担心的一个方面，因为我写的简单应用程序和库通常只在主线程上运行，或者不直接修改任何类中的属性或字段我以前需要担心。

但是，我已开始研究个人项目，我正在使用WebClient从远程服务器异步下载数据。有一个Queue<Uri>，它包含一系列下载数据的URI的预建队列。

因此考虑下面的代码片段（这不是我真正的代码，但有些事我希望说明我的问题：

private WebClient webClient = new WebClient(); 
private Queue<Uri> requestQueue = new Queue<Uri>(); 

public Boolean DownloadNextASync() 
{ 
    if (webClient.IsBusy) 
     return false; 

    if (requestQueue.Count == 0) 
     return false 

    var uri = requestQueue.Dequeue(); 

    webClient.DownloadDataASync(uri); 

    return true; 

} 

如果我理解正确，这种方法不是线程安全的（假设这个特定实例我的推理是webClient可能会在IsBusy检查和DownloadDataASync()方法调用之间的时间内变得繁忙，并且requestQueue可能在Count检查和下一个项目出列之间变为空时为

我的问题是处理这种类型的情况以使其线程安全的最佳方式是什么？

这是一个更抽象的问题，因为我认识到对于这种特定的方法，必须有一个非常不方便的时机才能真正引发问题，为了覆盖这种情况，我可以将该方法包装在合适的try-catch，因为这两件都会抛出异常。但是还有其他选择吗？ lock声明是否适用于此？

Answer 1

如果你的目标.NET 4.0，您可以使用任务并行库求助：

var queue = new BlockingCollection<Uri>(); 
var maxClients = 4; 

// Optionally provide another producer/consumer collection for the data 
// var data = new BlockingCollection<Tuple<Uri,byte[]>>(); 

// Optionally implement CancellationTokenSource 

var clients = from id in Enumerable.Range(0, maxClients) 
       select Task.Factory.StartNew(
    () => 
    { 
     var client = new WebClient(); 
     while (!queue.IsCompleted) 
     { 
      Uri uri; 
      if (queue.TryTake(out uri)) 
      { 
       byte[] datum = client.DownloadData(uri); // already "async" 
       // Optionally pass datum along to the other collection 
       // or work on it here 
      } 
      else Thread.SpinWait(100); 
     } 
    }); 

// Add URI's to search 
// queue.Add(...); 

// Notify our clients that we've added all the URI's 
queue.CompleteAdding(); 

// Wait for all of our clients to finish 
clients.WaitAll(); 

---

要使用这种方法取得进展指示您可以使用TaskCompletionSource<TResult>来管理基于事件的并行：

public static Task<byte[]> DownloadAsync(Uri uri, Action<double> progress) 
{ 
    var source = new TaskCompletionSource<byte[]>(); 
    Task.Factory.StartNew(
     () => 
     { 
      var client = new WebClient(); 
      client.DownloadProgressChanged 
       += (sender, e) => progress(e.ProgressPercentage); 
      client.DownloadDataCompleted 
       += (sender, e) => 
       { 
        if (!e.Cancelled) 
        { 
         if (e.Error == null) 
         { 
          source.SetResult((byte[])e.Result); 
         } 
         else 
         { 
          source.SetException(e.Error); 
         } 
        } 
        else 
        { 
         source.SetCanceled(); 
        } 
       }; 
     }); 

    return source.Task; 
} 

像这样使用：

// var urls = new List<Uri>(...); 
// var progressBar = new ProgressBar(); 

Task.Factory.StartNew(
    () => 
    { 
     foreach (var uri in urls) 
     { 
      var task = DownloadAsync(
       uri, 
       p => 
        progressBar.Invoke(
         new MethodInvoker(
         delegate { progressBar.Value = (int)(100 * p); })) 
       ); 

      // Will Block! 
      // data = task.Result; 
     } 
    }); 

Answer 2

我强烈推荐阅读Joseph Albahari的“Threading in C＃”。为了准备我的第一次（错误）线程冒险，我已经仔细研究了它，并且它非常全面。

你可以在这里阅读：http://www.albahari.com/threading/

Answer 3

你提出的两个线程安全问题都是有效的。此外，WebClient和Queue都被记录为不是线程安全的（位于MSDN文档的底部）。例如，如果两个线程同时出队，它们可能实际上会导致队列内部不一致或者可能导致非感性返回值。例如，如果出列（）的执行是这样的：

1. var valueToDequeue = this._internalList[this._startPointer]; 
2. this._startPointer = (this._startPointer + 1) % this._internalList.Count; 
3. return valueToDequeue; 

和两个线程各执行1号线之前，要么继续行2，然后（这里还有其他潜在的问题都将返回相同的值好）。这不一定会抛出异常，所以你应该使用lock语句，以保证只有一个线程可以在方法内部进行查询：

private readonly object _lock = new object(); 

... 

lock (this._lock) { 
    // body of method 
} 

你也可以锁定在Web客户端或队列，如果你知道没有人会同步他们。