从方法中将 BlockingCollection 作为 IEnumerable 返回

Question

我试图从 BlockingCollection 支持的方法返回 IEnumerable。代码模式是：

public IEnumerable<T> Execute() {   
    var results = new BlockingCollection<T>(10);  
    _ExecuteLoad(results);   
    return results.GetConsumingEnumerable(); 
}

private void _ExecuteLoad<T>(BlockingCollection<T> results) {
    var loadTask = Task.Factory.StartNew(() =>
    { 
        //some async code that adds items to results
        results.CompleteAdding();
    });
}

public void Consumer() {
    var count = Execute().Count();
}

问题是从 Execute() 返回的枚举始终为空。我见过的例子都是在任务中迭代BlockingCollection。在这种情况下似乎行不通。

有谁知道我哪里出错了？

---

为了让事情更清楚一些，我粘贴了正在执行的用于填充集合的代码。也许有什么东西导致了这里的问题？

Task.Factory.StartNew(() =>
{
    var continuationRowKey = "";
    var continuationParitionKey = "";
    var action = HttpMethod.Get;
    var queryUri = _GetTableQueryUri(tableServiceUri, tableName, query, continuationParitionKey, continuationRowKey, timeout);
    while (true)
    {
        using (var request = GetRequest(queryUri, null, action.Method, azureAccountName, azureAccountKey))
        {
            request.Method = action;
            request.RequestUri = queryUri;

            using (var client = new HttpClient())
            {
                var sendTask = client.SendAsync(request, HttpCompletionOption.ResponseHeadersRead);
                using (var response = sendTask.Result)
                {
                    continuationParitionKey = // stuff from headers
                    continuationRowKey = // stuff from headers

                    var streamTask = response.Content.ReadAsStreamAsync();
                    using (var stream = streamTask.Result)
                    {
                        using (var reader = XmlReader.Create(stream))
                        {
                            while (reader.Read())
                            {
                                if (reader.NodeType == XmlNodeType.Element && reader.Name == "entry" && reader.NamespaceURI == "http://www.w3.org/2005/Atom")
                                {
                                    results.Add(XNode.ReadFrom(reader) as XElement);
                                }
                            }
                            reader.Close();
                        }
                    }
                }
            }

            if (continuationParitionKey == null && continuationRowKey == null)
                break;

            queryUri = _GetTableQueryUri(tableServiceUri, tableName, query, continuationParitionKey, continuationRowKey, timeout);
        }
    }
    results.CompleteAdding();
});

Answer 1

当您完成向集合添加项目后，您需要调用results.CompleteAdding()。

如果不这样做，枚举将永远不会结束，并且 Count() 将永远不会返回。

除此之外，您发布的代码是正确的。