Red Hat Summit 红帽全球峰会支持控制台（Console）开发人员开始试用联系人

95.8. MongoDB operations - producer 端点

95.8.1. 查询操作

95.8.1.1. findById

此操作仅从与 IN 消息正文的内容匹配的集合中检索一个元素。传入的对象可以是等同于 Bson 类型的任何对象。请参阅 http://bsonspec.org/spec.html 和 http://www.mongodb.org/display/DOCS/Java+Types。 

from("direct:findById")
    .to("mongodb:myDb?database=flights&collection=tickets&operation=findById")
    .to("mock:resultFindById");

请注意，默认的 _id 被 Mongo 视为 和 ObjectId 类型，因此您可能需要正确转换它。 

from("direct:findById")
    .convertBodyTo(ObjectId.class)
    .to("mongodb:myDb?database=flights&collection=tickets&operation=findById")
    .to("mock:resultFindById");
注意

支持可选参数
此操作支持投射操作器。请参阅 指定字段过滤器（项目 ）。

95.8.1.2. findOneByQuery

从与 MongoDB 查询选择器匹配的集合中检索第一个元素。如果设置了 CamelMongoDbCriteria 标头，则其值将用作查询选择器。如果 CamelMongoDbCriteria 标头为 null，则 IN 消息正文将用作查询选择器。在这两种情况下，查询选择器都应是 Bson 类型，或者转换为 Bson （例如，JSON 字符串或 HashMap）。如需更多信息，请参阅类型转换。

使用 MongoDB Driver 提供的 过滤器 创建查询选择器。

95.8.1.3. 没有查询选择器的示例（返回集合中的第一个文档）

from("direct:findOneByQuery")
    .to("mongodb:myDb?database=flights&collection=tickets&operation=findOneByQuery")
    .to("mock:resultFindOneByQuery");

95.8.1.4. 使用查询选择器的示例（返回集合中的第一个匹配文档）：

from("direct:findOneByQuery")
    .setHeader(MongoDbConstants.CRITERIA, constant(Filters.eq("name", "Raul Kripalani")))
    .to("mongodb:myDb?database=flights&collection=tickets&operation=findOneByQuery")
    .to("mock:resultFindOneByQuery");
注意

支持可选参数
此操作支持投射运算符和排序条款。请参阅 指定字段过滤器（项目 ），指定 sort 子句。

95.8.1.5. findAll

findAll 操作会返回与查询匹配的所有文档，或者根本没有返回集合中包含的所有文档。查询对象提取 CamelMongoDbCriteria 标头。如果 CamelMongoDbCriteria 标头为 null，则查询对象被提取消息正文，即 Bson 类型或可转换为 Bson。它可以是 JSON 字符串或 Hashmap。如需更多信息，请参阅类型转换。

95.8.1.5.1. 没有查询选择器的示例（返回集合中的所有文档）
from("direct:findAll")
    .to("mongodb:myDb?database=flights&collection=tickets&operation=findAll")
    .to("mock:resultFindAll");
95.8.1.5.2. 带有查询选择器的示例（返回集合中的所有匹配文档）
from("direct:findAll")
    .setHeader(MongoDbConstants.CRITERIA, Filters.eq("name", "Raul Kripalani"))
    .to("mongodb:myDb?database=flights&collection=tickets&operation=findAll")
    .to("mock:resultFindAll");

通过以下标头支持分页和高效的检索：

标头键快速常量描述（从 MongoDB API 文档中提取）预期类型

CamelMongoDbNumToSkip

MongoDbConstants.NUM_TO_SKIP

在光标的开头丢弃给定数量的元素。

int/Integer

CamelMongoDbLimit

MongoDbConstants.LIMIT

限制返回的元素数量。

int/Integer

CamelMongoDbBatchSize

MongoDbConstants.BATCH_SIZE

限制一个批处理中返回的元素数量。光标通常获取结果对象的批处理，并将它们存储在本地。如果 batchSize 为正数，它代表检索的每个批处理对象的大小。可以调整它以优化性能并限制数据传输。如果 batchSize 为负数，它将限制返回的数量对象，该对象位于最大批处理大小限制（通常为 4MB），并且光标将关闭。例如，如果 batchSize 是 -10，那么服务器将返回最多 10 个文档，以及 4MB 中可以容纳的内容，然后关闭光标。请注意，这个功能与文档中的 limit （）不同，该文档中的大小必须符合最大大小，而且无需发送请求以关闭光标服务器端。即使光标迭代后也可以更改批处理大小，在这种情况下，设置将应用于下一个批处理检索。

int/Integer

CamelMongoDbAllowDiskUse

MongoDbConstants.ALLOW_DISK_USE

设置 allowDiskUse MongoDB 标志。从 MongoDB Server 4.3.1 开始，这被支持。将这个标头与旧的 MongoDB Server 版本搭配使用可能会导致查询失败。

布尔值/布尔值

95.8.1.5.3. 选项 outputType=MongoIterable 和 batch size 的示例
from("direct:findAll")
    .setHeader(MongoDbConstants.BATCH_SIZE).constant(10)
    .setHeader(MongoDbConstants.CRITERIA, constant(Filters.eq("name", "Raul Kripalani")))
    .to("mongodb:myDb?database=flights&collection=tickets&operation=findAll&outputType=MongoIterable")
    .to("mock:resultFindAll");

findAll 操作还会返回以下 OUT 标头，以便在使用分页时迭代结果页面：

标头键快速常量描述（从 MongoDB API 文档中提取）数据类型

CamelMongoDbResultTotalSize

MongoDbConstants.RESULT_TOTAL_SIZE

与查询匹配的对象数量。这不会考虑 limit/skip。

int/Integer

CamelMongoDbResultPageSize

MongoDbConstants.RESULT_PAGE_SIZE

与查询匹配的对象数量。这不会考虑 limit/skip。

int/Integer

注意

支持可选参数
此操作支持投射运算符和排序条款。请参阅 指定字段过滤器（项目 ），指定 sort 子句。

95.8.1.6. æ•°é‡�

返回集合中对象总数，返回 Long 作为 OUT 消息正文。
以下示例将计算 "dynamicCollectionName" 集合中记录的数量。注意如何启用动态性，因此操作不会针对"notableScientists"集合运行，而是针对"dynamicCollectionName"集合运行。 

// from("direct:count").to("mongodb:myDb?database=tickets&collection=flights&operation=count&dynamicity=true");
Long result = template.requestBodyAndHeader("direct:count", "irrelevantBody", MongoDbConstants.COLLECTION, "dynamicCollectionName");
assertTrue("Result is not of type Long", result instanceof Long);

您可以提供 查询对象提取 CamelMongoDbCriteria 标头。如果 CamelMongoDbCriteria 标头为 null，则查询对象被提取的消息正文，即 Bson 或 convertable to Bson。 

Document query = ...
Long count = template.requestBodyAndHeader("direct:count", query, MongoDbConstants.COLLECTION, "dynamicCollectionName");

95.8.1.7. 指定字段过滤器（项目）

默认情况下，查询操作将在其整个中返回匹配的对象（及其所有字段）。如果您的文档非常大，且您只需要检索其字段的子集，您可以在所有查询操作中指定字段过滤器，只需通过设置相关的 Bson （或 typeible to Bson，如 JSON String、Map 等）在 CamelMongoDbFieldsProjection 标头中指定字段过滤器。

以下是一个示例，它使用 MongoDB 的 Projections 来简化 Bson 的创建。它检索除 _id 和 boringField 以外的所有字段： 

// route: from("direct:findAll").to("mongodb:myDb?database=flights&collection=tickets&operation=findAll")
Bson fieldProjection = Projection.exclude("_id", "boringField");
Object result = template.requestBodyAndHeader("direct:findAll", ObjectUtils.NULL, MongoDbConstants.FIELDS_PROJECTION, fieldProjection);

以下是一个示例，它使用 MongoDB 的 Projections 来简化 Bson 的创建。它检索除 _id 和 boringField 以外的所有字段： 

// route: from("direct:findAll").to("mongodb:myDb?database=flights&collection=tickets&operation=findAll")
Bson fieldProjection = Projection.exclude("_id", "boringField");
Object result = template.requestBodyAndHeader("direct:findAll", ObjectUtils.NULL, MongoDbConstants.FIELDS_PROJECTION, fieldProjection);

95.8.1.8. 指定 sort 子句

通常，根据特定字段排序从集合中获取 min/max 记录的要求，该字段使用 MongoDB 的 排序来简化 Bs on 的创建。它检索除 _id 和 boringField 以外的所有字段： 

// route: from("direct:findAll").to("mongodb:myDb?database=flights&collection=tickets&operation=findAll")
Bson sorts = Sorts.descending("_id");
Object result = template.requestBodyAndHeader("direct:findAll", ObjectUtils.NULL, MongoDbConstants.SORT_BY, sorts);

在 Camel 路由中，SORT_BY 标头可用于 findOneByQuery 操作，以实现相同的结果。如果还指定了 FIELDS_PROJECTION 标头，则操作将返回单个字段/值对，可直接传递给另一个组件（例如，参数ized MyBatis SELECT 查询）。本例演示了如何根据 documentTimestamp 字段从集合中获取临时最新的文档，并将结果减少到单个字段： 

.from("direct:someTriggeringEvent")
.setHeader(MongoDbConstants.SORT_BY).constant(Sorts.descending("documentTimestamp"))
.setHeader(MongoDbConstants.FIELDS_PROJECTION).constant(Projection.include("documentTimestamp"))
.setBody().constant("{}")
.to("mongodb:myDb?database=local&collection=myDemoCollection&operation=findOneByQuery")
.to("direct:aMyBatisParameterizedSelect");

95.8.2. 创建/更新操作

95.8.2.1. insert

将新对象插入到 MongoDB 集合中，从 IN 消息正文获取。试图将类型转换转换为 文档或 列表。
支持两种模式：单一插入和多个插入。对于多个插入，端点将预期任意类型的对象列表、数组或集合，只要它们是 - 或可以转换为 - 文档。Example: 

from("direct:insert")
    .to("mongodb:myDb?database=flights&collection=tickets&operation=insert");

该操作将返回 WriteResult，具体取决于 WriteConcern 或 invokeGetLastError 选项的值，getLastError （） 将被调用。如果您希望访问些操作的最终结果，可以通过在 WriteResult 上调用 getLastError() 或 getCachedLastError() 来获取 CommandResult。然后，您可以通过调用 CommandResult.ok （）, CommandResult.getErrorMessage （） and/or CommandResult.getException （） 来验证结果。

请注意，新对象的 _id 在集合中必须是唯一的。如果没有指定值，MongoDB 将自动为您生成一个值。但是，如果您确实指定了它，则插入操作将失败（对于 Camel 需要注意，您需要启用 invokeGetLastError 或设置等待写结果的 WriteConcern）。

这并不是组件的限制，而是在 MongoDB 中的工作方式以获得更高的吞吐量。如果您使用自定义 _id，您预期在应用程序级别确保是唯一的（这也是很好的做法）。

插入记录的 OID 存储在 CamelMongoOid 键下的消息标头中(MongoDbConstants.OID constant)。存储的值为 org.bson.types.ObjectId，用于单个插入或 java.util.List<org.bson.types.ObjectId&gt; （如果插入多个记录）。

在 MongoDB Java Driver 3.x 中，insertOne 和 insertMany 操作返回 void。Camel 插入操作会返回插入的 Documents 或 Documents 列表。请注意，如果需要，每个文档都由一个新的 OID 更新。

95.8.2.2. save

save 操作等同于 upsert (UPdate，inSERT)操作，其中将更新记录，如果不存在，它将被插入一个原子操作。MongoDB 将根据 _id 字段执行匹配。

请注意，如果更新，对象会被完全被替换，而不允许使用 MongoDB 的 $modifier。因此，如果要操作对象已存在，则有两个选项：

  1. 执行查询以首先检索整个对象及其所有字段（或者效率不高），在 Camel 中更改它，然后保存它。
  2. 使用带有 $modifiers 的 update 操作，这将在服务器端执行更新。您可以启用 upsert 标志，在这种情况下，如果需要插入，MongoDB 会将 $modifiers 应用到过滤器查询对象并插入结果。

如果要保存的文档不包含 _id 属性，则操作将是插入的，并且创建新的 _id 将放在 CamelMongoOid 标头中。

例如： 

from("direct:insert")
    .to("mongodb:myDb?database=flights&collection=tickets&operation=save");
// route: from("direct:insert").to("mongodb:myDb?database=flights&collection=tickets&operation=save");
org.bson.Document docForSave = new org.bson.Document();
docForSave.put("key", "value");
Object result = template.requestBody("direct:insert", docForSave);

95.8.2.3. update

更新集合上的一个或多个记录。需要过滤器查询和更新规则。

您可以使用 MongoDBConstants.CRITERIA 标头定义为 Bson，并在 Body 中将更新规则定义为 Bson。

注意

在功能改进
后，使用 MongoDBConstants.CRITERIA 标头定义为 Bson 在进行更新前查询 mongodb，但您应该注意，在聚合中使用增强模式并应用 mongodb 更新，则需要在聚合过程中将其从生成的 camel Exchange 中删除，然后应用 mongodb 更新。如果您在聚合和/或重新定义 MongoDBConstants.CRITERIA 标头前没有删除这个标头，在将 camel Exchange 发送到 mongodb producer 端点前，您可能会在更新 mongodb 时最终使用无效的 camel Exchange payload。

第二种方法 Require a List<Bson> 作为 IN message body contains exactly 2 项：

  • 元素 1 (index 0) jpeg 过滤器查询 TOKEN 确定受影响的对象，与典型的查询对象相同
  • 元素 2 (index 1) TOKEN 更新规则 jpeg 如何更新匹配的对象。支持 MongoDB 中的所有 修饰符操作。
注意

Multiupdates
默认情况下，MongoDB 只会更新 1 个对象，即使多个对象与过滤器查询匹配。要指示 MongoDB 更新 所有匹配 记录，请将 CamelMongoDbMultiUpdate IN 消息标头设置为 true。

将返回一个带有键 CamelMongoDbRecordsAffected 的标头(MongoDbConstants.RECORDS_AFFECTED constant)，其记录数量已更新（从 WriteResult.getN （））。

支持以下 IN 消息标头：

标头键快速常量描述（从 MongoDB API 文档中提取）预期类型

CamelMongoDbMultiUpdate

MongoDbConstants.MULTIUPDATE

如果更新应该应用到所有匹配的对象。请参阅 http://www.mongodb.org/display/DOCS/Atomic+Operations

布尔值/布尔值

CamelMongoDbUpsert

MongoDbConstants.UPSERT

如果数据库应该创建元素（如果不存在）

布尔值/布尔值

例如，以下命令将 "scientist" 字段的值设置为 "Darwin" 来更新其 filterField 字段等于 true 的所有记录： 

// route: from("direct:update").to("mongodb:myDb?database=science&collection=notableScientists&operation=update");
List<Bson> body = new ArrayList<>();
Bson filterField = Filters.eq("filterField", true);
body.add(filterField);
BsonDocument updateObj = new BsonDocument().append("$set", new BsonDocument("scientist", new BsonString("Darwin")));
body.add(updateObj);
Object result = template.requestBodyAndHeader("direct:update", body, MongoDbConstants.MULTIUPDATE, true);
// route: from("direct:update").to("mongodb:myDb?database=science&collection=notableScientists&operation=update");
Maps<String, Object> headers = new HashMap<>(2);
headers.add(MongoDbConstants.MULTIUPDATE, true);
headers.add(MongoDbConstants.FIELDS_FILTER, Filters.eq("filterField", true));
String updateObj = Updates.set("scientist", "Darwin");;
Object result = template.requestBodyAndHeaders("direct:update", updateObj, headers);
// route: from("direct:update").to("mongodb:myDb?database=science&collection=notableScientists&operation=update");
String updateObj = "[{\"filterField\": true}, {\"$set\", {\"scientist\", \"Darwin\"}}]";
Object result = template.requestBodyAndHeader("direct:update", updateObj, MongoDbConstants.MULTIUPDATE, true);

95.8.3. 删除操作

95.8.3.1. remove

从集合中删除匹配的记录。IN 消息正文将充当删除过滤器查询，它预期为 DBObject 类型或可转换的类型。
以下示例将删除在 Science 数据库, notableScientists 集合中，其字段 'conditionField' 等于 true 的所有对象： 

// route: from("direct:remove").to("mongodb:myDb?database=science&collection=notableScientists&operation=remove");
Bson conditionField = Filters.eq("conditionField", true);
Object result = template.requestBody("direct:remove", conditionField);

返回一个带有键 CamelMongoDbRecordsAffected 的标头 (MongoDbConstants.RECORDS_AFFECTED constant) with type int, 包括删除记录的数量 (从 WriteResult.getN() 复制)。

95.8.4. 批量写操作

95.8.4.1. bulkWrite

批量执行写入操作，并控制执行顺序。需要 List<WriteModel<Document >> 作为 IN 消息正文，其中包含用于插入、更新和删除操作的命令。

以下示例将插入一个新的科学家 "Pierre Curie"，更新 id 为 "5" 的记录，将 "scientist" 项的值设置为 "Marie Curie" 并删除 id 为 "3" 的记录 : 

// route: from("direct:bulkWrite").to("mongodb:myDb?database=science&collection=notableScientists&operation=bulkWrite");
List<WriteModel<Document>> bulkOperations = Arrays.asList(
            new InsertOneModel<>(new Document("scientist", "Pierre Curie")),
            new UpdateOneModel<>(new Document("_id", "5"),
                                 new Document("$set", new Document("scientist", "Marie Curie"))),
            new DeleteOneModel<>(new Document("_id", "3")));

BulkWriteResult result = template.requestBody("direct:bulkWrite", bulkOperations, BulkWriteResult.class);

默认情况下，操作会按顺序执行，并在第一个写入错误上中断，而不处理列表中任何剩余的写入操作。要指示 MongoDB 继续处理列表中剩余的写入操作，请将 CamelMongoDbBulkOrdered IN 消息标头设置为 false。未排序的操作是并行执行的，无法保证此行为。

标头键快速常量描述（从 MongoDB API 文档中提取）预期类型

CamelMongoDbBulkOrdered

MongoDbConstants.BULK_ORDERED

执行有序或未排序的操作执行。默认值为 true。

布尔值/布尔值

95.8.5. 其他操作

95.8.5.1. 聚合

使用正文中包含的给定管道执行聚合。聚合可能非常长且繁重的操作。请谨慎使用。 

// route: from("direct:aggregate").to("mongodb:myDb?database=science&collection=notableScientists&operation=aggregate");
List<Bson> aggregate = Arrays.asList(match(or(eq("scientist", "Darwin"), eq("scientist",
        group("$scientist", sum("count", 1)));
from("direct:aggregate")
    .setBody().constant(aggregate)
    .to("mongodb:myDb?database=science&collection=notableScientists&operation=aggregate")
    .to("mock:resultAggregate");

支持以下 IN 消息标头：

标头键快速常量描述（从 MongoDB API 文档中提取）预期类型

CamelMongoDbBatchSize

MongoDbConstants.BATCH_SIZE

设置每个批处理要返回的文档数量。

int/Integer

CamelMongoDbAllowDiskUse

MongoDbConstants.ALLOW_DISK_USE

启用聚合管道阶段，将数据写入临时文件。

布尔值/布尔值

默认情况下，返回所有结果的列表。根据结果的大小，这在内存上可能很重。一个安全的替代方案是设置 outputType=MongoIterable。下一个处理器将在消息正文中看到可迭代的结果，以便它逐一完成结果。因此，设置批处理大小并返回一个可迭代功能，以便有效地检索和处理结果。

一个示例类似如下： 

List<Bson> aggregate = Arrays.asList(match(or(eq("scientist", "Darwin"), eq("scientist",
        group("$scientist", sum("count", 1)));
from("direct:aggregate")
    .setHeader(MongoDbConstants.BATCH_SIZE).constant(10)
    .setBody().constant(aggregate)
    .to("mongodb:myDb?database=science&collection=notableScientists&operation=aggregate&outputType=MongoIterable")
    .split(body())
    .streaming()
    .to("mock:resultAggregate");

请注意，调用 .split (body （)） 足以发送路由一对条目，但仍然会首先将所有条目加载到内存中。因此，需要调用 .streaming （） 来通过批处理将数据加载到内存中。

95.8.5.2. getDbStats

等同于在 MongoDB shell 中运行 db.stats （） 命令，该命令显示有关数据库的有用统计图。
例如： 

> db.stats();
{
    "db" : "test",
    "collections" : 7,
    "objects" : 719,
    "avgObjSize" : 59.73296244784423,
    "dataSize" : 42948,
    "storageSize" : 1000058880,
    "numExtents" : 9,
    "indexes" : 4,
    "indexSize" : 32704,
    "fileSize" : 1275068416,
    "nsSizeMB" : 16,
    "ok" : 1
}

使用示例： 

// from("direct:getDbStats").to("mongodb:myDb?database=flights&collection=tickets&operation=getDbStats");
Object result = template.requestBody("direct:getDbStats", "irrelevantBody");
assertTrue("Result is not of type Document", result instanceof Document);

该操作将返回与 shell 中显示的数据结构，格式为 OUT 消息正文中的 Document 形式。

95.8.5.3. getColStats

等同于在 MongoDB shell 中运行 db.collection.stats （） 命令，该命令显示有关集合的有用统计图。
例如： 

> db.camelTest.stats();
{
    "ns" : "test.camelTest",
    "count" : 100,
    "size" : 5792,
    "avgObjSize" : 57.92,
    "storageSize" : 20480,
    "numExtents" : 2,
    "nindexes" : 1,
    "lastExtentSize" : 16384,
    "paddingFactor" : 1,
    "flags" : 1,
    "totalIndexSize" : 8176,
    "indexSizes" : {
        "_id_" : 8176
    },
    "ok" : 1
}

使用示例： 

// from("direct:getColStats").to("mongodb:myDb?database=flights&collection=tickets&operation=getColStats");
Object result = template.requestBody("direct:getColStats", "irrelevantBody");
assertTrue("Result is not of type Document", result instanceof Document);

该操作将返回与 shell 中显示的数据结构，格式为 OUT 消息正文中的 Document 形式。

95.8.5.4. 命令

作为命令在数据库上运行正文。在获取主机信息、复制或分片状态时，对 admin 操作很有用。

这个操作不使用 collection 参数。 

// route: from("command").to("mongodb:myDb?database=science&operation=command");
DBObject commandBody = new BasicDBObject("hostInfo", "1");
Object result = template.requestBody("direct:command", commandBody);

95.8.6. 动态操作

Exchange 可以通过设置由 MongoDbConstants.OPERATION_HEADER 常数定义的 CamelMongoDbOperation 标头来覆盖端点的固定操作。
支持的值由 MongoDbOperation enumeration 决定，并与端点 URI 上的 operation 参数接受的值匹配。

例如： 

// from("direct:insert").to("mongodb:myDb?database=flights&collection=tickets&operation=insert");
Object result = template.requestBodyAndHeader("direct:insert", "irrelevantBody", MongoDbConstants.OPERATION_HEADER, "count");
assertTrue("Result is not of type Long", result instanceof Long);
Red Hat logoGithubRedditYoutubeTwitter

学习

尝试、购买和销售

社区

关于红帽文档

通过我们的产品和服务，以及可以信赖的内容，帮助红帽用户创新并实现他们的目标。

让开源更具包容性

红帽致力于替换我们的代码、文档和 Web 属性中存在问题的语言。欲了解更多详情，请参阅红帽博客.

關於紅帽

我们提供强化的解决方案，使企业能够更轻松地跨平台和环境（从核心数据中心到网络边缘）工作。

© 2024 Red Hat, Inc.