c# - 从可以重新添加对象的阻塞集合中消费

Question

所以我使用阻塞集合来存储自定义类 Node，它将保存一个数据库连接和一个在该连接上运行的查询。集合在被消费之前就完成了。但是，如果我从集合中拉出一个节点并尝试运行它，它可能会失败，我想将它重新添加到集合中以便稍后重新运行。

我有两个可行的解决方案，但我都不喜欢其中任何一个，希望有人能给我一个更简洁的解决方案，或者一些关于如何改进它的想法。

第一:集合上的 Parallel.Foreach，任何失败的内容都会添加到递归调用的新阻塞集合中。

Parallel.ForEach(NodeList, node => {
    try {
        using (NpgsqlConnection conn = new NpgsqlConnection(node.ConnectionString)) {
            conn.Open();
            using (NpgsqlCommand npgQuery = new NpgsqlCommand(node.Query, conn)) {
                using (NpgsqlDataReader reader = npgQuery.ExecuteReader()) {
                    while (reader.Read()) {
                        //Do stuff
                    }
                }
            }
        }
    } catch (Exception e){
            retryNodes.Add(node);
    }
});
retryNodes.CompleteAdding();
NodeList = retryNodes.ToList<Node>();
try {
    ExecuteNodes();
} catch (Exception e) {
    throw e;
}

我不喜欢这样，因为这意味着当原始集合结束时，它正在浪费线程等待新集合开始。

第二:为集合中的每个项目手动启动任务:

int totalToFinish = NodeList.Count;
while (totalToFinish > 0) {
    while (threadsRunning < MaxAllowedThreads && totalToFinish > 0) {
        Interlocked.Increment(ref threadsRunning);
        Task.Factory.StartNew(() => {
            if (NodeList.Count == 0) {
                Interlocked.Decrement(ref threadsRunning);
                return;
            }
            Node node;
            NodeList.TryTake(out node, 1000);
            if (node.Attempts++ >= RetryAttempts) {
                throw new Exception("Failed after allowed attemps of: " + RetryAttempts);
            }
            try {
                using (NpgsqlConnection conn = new NpgsqlConnection(node.ConnectionString)) {
                    conn.Open();
                    using (NpgsqlCommand npgQuery = new NpgsqlCommand(node.Query, conn)) {
                        using (NpgsqlDataReader reader = npgQuery.ExecuteReader()) {
                            while (reader.Read()) {
                                //Do stuff
                            }
                        }
                    }
                    Interlocked.Decrement(ref totalToFinish);
                }
            } catch (Exception e) {
                NodeList.Add(node);
            }
            Interlocked.Decrement(ref threadsRunning);
        });
    }
}

这种方式在性能方面效果更好，但开销很大，我觉得这不是一个好方法。

如果有人能帮助我，我将不胜感激。

谢谢。

Answer 1

我看到您在第二种算法中设置了重试次数限制 - 您可以通过包含此重试循环来简化第一种算法

Parallel.ForEach(NodeList, node => {
    while(true) {
        try {
            using (NpgsqlConnection conn = new NpgsqlConnection(node.ConnectionString)) {
                conn.Open();
                using (NpgsqlCommand npgQuery = new NpgsqlCommand(node.Query, conn)) {
                    using (NpgsqlDataReader reader = npgQuery.ExecuteReader()) {
                        while (reader.Read()) {
                            //Do stuff
                        }
                    }
                }
            }
            break; // break out of outer while loop
        } catch (Exception e){
            node.Attempts++;
            if(node.Attempts >= RetryAttempts) {
                throw new Exception("Too many retries");
            }
        }
    }
});