MongoDB 和多个 upsert答案

【问题标题】：MongoDB and multiple upsertMongoDB 和多个 upsert
【发布时间】：2013-10-30 21:20:01
【问题描述】：

我对 MongoDB 比较陌生，但我们考虑将其用作遗留服务前的某种缓存。在这种情况下，我们偶然发现了一些问题。

首先，一些解释。

此缓存服务将位于旧服务和客户端之间。客户端将连接到缓存服务，该服务从旧服务获取数据。缓存服务每 X 分钟获取一次数据，并将它们保存在 MongoDB 中。架构真的很简单：只是一个包含很多键/值的文档。没有嵌套文档等。此外，我们将 _id 设置为来自旧服务的唯一 ID，因此我们也可以控制它。

当缓存服务从旧服务获取数据时，它只获得一个增量（仅自上次获取后发生变化）。因此，如果自上次以来有 5 个“对象”发生了变化，那么您只会得到这 5 个“对象”（但您会得到完整的对象，而不是对象的增量）。如果任何新的“对象”已添加到旧版服务中，那么这些对象当然也在 delta 版中。

我们的“问题”

在我看来，这听起来像是一个更新。如果有新对象，请插入它们。如果现有对象发生更改，请更新它们。然而，MongoDB 似乎并不特别喜欢多个 upsert。只需插入就会给我一个关于重复键的错误，这是完全可以理解的，因为已经存在具有相同 _id 的文档。可以接受 upsert 参数的更新函数不能接受新对象的列表。在我看来，单个查询是不可能的。不过，我也有可能在这里完全忽略了某些东西。

可能的解决方案

有许多不同的解决方案，尤其是我想到的两个：

执行两个查询：首先，计算一个包含所有 _id 的列表（请记住，我们从旧服务中获得这些）。然后，使用 $in 函数连同 _id 列表删除它们，并立即插入新文档。这实际上应该用新数据更新我们的集合。它也很容易实现。可能出现的一个问题是客户端在删除和插入之间请求数据，因此错误地得到一个空结果。这会破坏交易，绝对不会发生。
对每个更改的对象执行一次更新插入。也很容易实现，并且不应该给出与其他解决方案相同的问题。不过，这还有其他（可能是想象的）问题。它可以在短时间内处理多少个 upsert？它每分钟可以轻松处理 5000 个 upsert 吗？这些不是大文档，只有大约 20 个键/值，没有子文档。这个数字是凭空捏造的，实际数字很难预测。在我看来，这种方法感觉不对。我不明白为什么必须为每个新文档运行一个查询。

我们将不胜感激任何关于这两个提议的解决方案和任何其他解决方案的帮助。作为旁注，技术不是真正可讨论的，所以请不要建议其他类型的数据库或语言。我们选择我们选择的东西还有其他强有力的原因:)

【问题讨论】：

标签： java mongodb caching mongodb-java

【解决方案1】：

我会分享我的经验...

在我上一份工作中，我们也遇到过类似的情况。我们最终对每个文档/对象进行了一次查询/写入。我们使用 Mule ESB 将数据从遗留系统泵送到 Mongo，每次写入都是一次 upsert。

表现不错，但不是很好。我们可以在几分钟内将数千个文档导入 Mongo。这些文档相当丰富，因此这可能是我们不得不限制对 Mongo 的写入的部分原因。

在我们批量加载数据后，“实时”性能不再是问题。

您建议的第一个选项听起来太复杂了，并且可能会使 Mongo 处于未知状态，以防操作在更新中途终止。 upsert 选项为我们节省了很多时间，因为我们可以一遍又一遍地重播插入并且是安全的。

【讨论】：

感谢您的回答。我刚刚测试过，插入对象确实非常快。稍后我将不得不进行更多测试，但到目前为止，这似乎是一个很有前途的解决方案:)

【解决方案2】：

扩展 ryan1234 的回答：

2.6 版本的 MongoDB 将具备发送批量更新的能力。现在，您需要为每个文档提交单独的请求。

正如 ryan1234 所说，如果您不知道旧版提供者的信息，那么对每个文档进行 upsert 是更新所有现有文档并添加新文档的唯一安全方法。单个 MongoDB 进程可以在中端硬件上轻松处理每秒数千次更新 (1)。如果您没有获得该级别的性能，那么可能是客户端和 MongDB 服务器之间的请求延迟。 Asynchronous Java Driver 可以通过允许同时向服务器发送多个更新请求来帮助克服这一限制，同时将客户端复杂性/线程降至最低。

HTH，罗伯

1：我假设文档没有增长，也没有索引更新，但即使有了这些，您也应该能够每秒进行一千次更新。

【讨论】：

谢谢！我会看看 2.6 版本，看看它是否对性能有任何影响:)

【解决方案3】：

或者如果你的键是复合的，你可以使用：

public static BulkWriteResult insertAll(MongoCollection<Document> coll, List<Document> docs, String[] keyTags, boolean upsert) {
    if(docs.isEmpty())
        return null;
    List<UpdateOneModel<Document>> requests = new ArrayList<>(docs.size());
    UpdateOptions opt = new UpdateOptions().upsert(upsert);
    for (Document doc : docs ) {
        BasicDBObject filter = new BasicDBObject();
        for (String keyTag : keyTags) {
            filter.append(keyTag, doc.get(keyTag));
        }
        BasicDBObject action = new BasicDBObject("$set", doc);
        requests.add(new UpdateOneModel<Document>(filter, action, opt));
    }
    return coll.bulkWrite(requests);
}

【讨论】：

【解决方案4】：

我知道。它必须深入挖掘才能找到正确的方法。试试这个： /** * 将文档中的所有项目插入到集合中。 * @param coll 目标集合 * @param docs 新的或更新的文档 * @param keyTag 文档中键的名称 * @param upsert if true 如果没有找到则创建一个新文档 * @return BulkWriteResult 或 null 如果 docs.isEmpty() */

    public static BulkWriteResult insertAll(MongoCollection<Document> coll, List<Document> docs, String keyTag, boolean upsert) {
    if(docs.isEmpty())
        return null;
    List<UpdateOneModel<Document>> requests = new ArrayList<>(docs.size());
    UpdateOptions opt = new UpdateOptions().upsert(upsert);
    for (Document doc : docs ) {
        BasicDBObject filter = new BasicDBObject(keyTag, doc.get(keyTag)); 
        BasicDBObject action = new BasicDBObject("$set", doc);
        requests.add(new UpdateOneModel<Document>(filter, action, opt));
    }
    return coll.bulkWrite(requests);
}

【讨论】：