多线程NpgsqlConnections和阅读器产生重复的结果。C#

Question

我需要在10个数据库的数千个不同模式上运行数千个查询。我试图执行这些查询，并使用BlockingCollection将结果写入其中，同时使用另一个线程从该集合读取并将其写入磁盘，因为这些查询的结果集太大，无法存储在内存中。

下面是我代码中的问题区域：

public class Node {
    public string ConnectionString;
    public string Query;

    public Node(string databaseDetails, string query) {
        //Cannot put in actual logic, but this part is fine
        ConnectionString = {logic for connection string}
        Query = "set search_path to {schema from databaseDetails};" + query
    }
}

public void runQuery(string query, BlockingCollection<Dictionary<string, object>> producer) {
        List<Node> nodes = getNodes(query);
        Parallel.ForEach(nodes, node => {
            NpgsqlConnection conn = new NpgsqlConnection(node.ConnectionString);
            conn.Open();
            NpgsqlCommand npgQuery = new NpgsqlCommand(node.Query, conn);
            NpgsqlDataReader reader = npgQuery.ExecuteReader();
            while (reader.Read()) {
                Dictionary<string, object> row = new Dictionary<string, object>();
                for (int i = 0; i < reader.FieldCount; i++) {
                    row[reader.GetName(i)] = reader.GetValue(i);
                }
                producer.Add(row);
            }
            conn.Close();
        });
        producer.CompleteAdding();
}

这段代码运行并检索所有的结果，但是它也复制了很多结果，所以阻塞集合的记录比它应该的多5-10倍。任何帮助都会得到极大的重视。

Answer 1

所以我只是个白痴，把我生成的结果集和我正在运行的所有查询(而不是UNION )进行比较，所以我的“真”结果集中没有重复，因为联合正在删除它们:/ /