[MongoDB Java 异步驱动程序快速指南
导读
mongodb-java-driver是mongodb的Java驱动项目。
本文是对MongoDB-java-driver官方文档 MongoDB Async Driver Quick Tour 的翻译(原创翻译)。
mongodb-java-driver 从3.0版本开始同时支持同步、异步方式(分别是不同的驱动应用)。异步的好处,众所周知,就是支持快速、非阻塞式的IO操作,可以提高处理速度。
请注意:本文仅介绍异步驱动的使用指南。同步驱动官方文档:mongo-java-driver ,需要了解的朋友,请移驾。
安装
简单提下安装说明。
注:MongoDB 异步驱动需要依赖Netty 或 Java 7。
如果你的项目是maven项目,只需在pom.xml中添加如下依赖:
<dependencies>
<dependency>
<groupId>org.mongodb</groupId>
<artifactId>mongodb-driver-async</artifactId>
<version>3.3.0</version>
</dependency>
</dependencies>
你也可以点击链接直接下载jar包: 下载点这里 。
分割线,下面是 MongoDB Async Driver Quick Tour 的译文。
MongoDB 异步驱动快速指南
以下的代码片段来自于 async driver source 的范例代码 QuickTour.java
。
注意
如何安装MongoDB异步驱动请参考 安装指导 。
执行异步回调
MongoDB异步驱动利用Netty或Java7的AsynchronousSocketChannel
来提供一个支持异步的API,以支持快速的、非阻塞式的IO操作。
该API形式和MongoDB同步驱动的新API保持一致,但是任何会导致网络IO的方法都会有一个SingleResponseCallback
并且会立即返回,其中T
是响应对于该文档的类型的任何方法。
SingleResponseCallback
回调接口需要实现一个简单方法onResult(T result, Throwable t)
,这个方法在操作完成时被调用。其中,如果操作成功, result
参数包含着操作结果;如果操作失败,t
中包含着抛出的异常信息。
重要
在SingleResponseCallback
的实现中检查错误并适当处理错误是十分重要的。下面的错误检查仅为简便起见而省略。
创建一个连接
下面的例子展示多种方法去链接本地机器上的mydb
数据库。详情参考 MongoClients.create API手册。
// 直接连接默认服务host和端口,即 localhost:27017
MongoClient mongoClient = MongoClients.create();
// 使用一个字符串
MongoClient mongoClient = MongoClients.create("mongodb://localhost");
// 使用一个ConnectionString
MongoClient mongoClient = MongoClients.create(new ConnectionString("mongodb://localhost"));
// 使用MongoClientSettings
ClusterSettings clusterSettings = ClusterSettings.builder().hosts(asList(new ServerAddress("localhost"))).build();
MongoClientSettings settings = MongoClientSettings.builder().clusterSettings(clusterSettings).build();
MongoClient mongoClient = MongoClients.create(settings);
MongoDatabase database = mongoClient.getDatabase("mydb");
此时,database
对象是一个MongoDB 服务器中指定数据库的连接。
注意
getDatabase("mydb")
方法并没有回调,因为它没有涉及网络IO操作。一个 MongoDatabase
实例提供了与数据库进行交互的方法,若数据库不存在,它会在插入数据时创建一个新的数据库。例如,创建一个 collection 或插入 document(这些确实需要回调,因为需要涉及网络IO)。
MongoClient
MongoClient
实例实际上代表了一个数据库的连接池;即使要并发执行异步操作,你也仅仅需要一个 MongoClient
实例。
重要
一般情况下,在一个指定的数据库集群中仅需要创建一个MongoClient
实例,并通过你的应用使用它。
当创建多个实例时:
- 所有的资源使用限制(例如最大连接数)适用于每个
MongoClient
实例 - 销毁实例时,请确保调用
MongoClient.close()
清理资源。
获得一个 collection
要获得一个 collection ,你需要在 getCollection(String collectionName) 方法中指定 collection 的名字:
下面的例子获得名为 test
的collection :
MongoCollection<Document> collection = database.getCollection("test");
添加一个 document
一旦你有了collection对象,你就可以向collection中插入document。例如,考虑如下的json形式document;document中含包含了一个名为 info
的子document。
{
"name" : "MongoDB",
"type" : "database",
"count" : 1,
"info" : {
x : 203,
y : 102
}
}
要创建document,需要使用 Document 类。你可以使用这个类来创建嵌入式的document。
Document doc = new Document("name", "MongoDB")
.append("type", "database")
.append("count", 1)
.append("info", new Document("x", 203).append("y", 102));
要向 collection 中插入 document ,需要使用 insertOne()
方法。
collection.insertOne(doc, new SingleResultCallback<Void>() {
@Override
public void onResult(final Void result, final Throwable t) {
System.out.println("Inserted!");
}
});
SingleResponseCallback
是一个 函数式接口 并且它可以以lambda方式实现(前提是你的APP工作在JDK8):
collection.insertOne(doc, (Void result, final Throwable t) -> System.out.println("Inserted!"));
一旦document成功插入,onResult 回调方法会被调用并打印“Inserted!”。记住,在一个普通应用中,你应该总是检查 t
变量中是否有错误信息。
添加多个 document
要添加多个 documents,你可以使用 insertMany()
方法。
接下来的例子会添多个document,document形式如下:
{ "i" : value }
循环创建多个 documents 。
List<Document> documents = new ArrayList<Document>();
for (int i = 0; i < 100; i++) {
documents.add(new Document("i", i));
}
要插入多个 document 到 collection,传递 documents 列表到 insertMany()
方法.
collection.insertMany(documents, new SingleResultCallback<Void>() {
@Override
public void onResult(final Void result, final Throwable t) {
System.out.println("Documents inserted!");
}
});
统计一个 collection的document数量
既然前面的多个例子中我们已经插入了 101 个 document,我们可以检查一下插入数量,使用 count() 方法。下面的代码应该打印 101
。
collection.count(
new SingleResultCallback<Long>() {
@Override
public void onResult(final Long count, final Throwable t) {
System.out.println(count);
}
});
查询 collection
使用 find() 方法来查询 collection。
在一个 collection 中找到第一个 document
要获得 collection 中的第一个 document ,需要调用 first() 方法。collection.find().first()
返回第一个 document 或 null 值,而不是一个游标。这种查询适用于匹配一个单一的 document,,或你仅对第一个 document 有兴趣。
注意
有时你需要多次使用相同或相似的回调方法。在这种情况下,合理的做法是DRY(不要重复自己):把回调保存为一个具体的类或分配给一个变量。
SingleResultCallback<Document> printDocument = new SingleResultCallback<Document>() {
@Override
public void onResult(final Document document, final Throwable t) {
System.out.println(document.toJson());
}
};
下面的例子传递 printDocument
回调给 first
方法:
collection.find().first(printDocument);
范例会打印下面的 document:
{ "_id" : { "$oid" : "551582c558c7b4fbacf16735" },
"name" : "MongoDB", "type" : "database", "count" : 1,
"info" : { "x" : 203, "y" : 102 } }
注意
_id
元素会被MongoDB动态的添加到你的 document 上,并且值也会与展示的不同。“_” 和 “$”开头的域是MongoDB 预留给内部使用的。
遍历查找一个collection中所有的 document
要检索 collection 中所有的 document,需要使用 find()
方法。find() 方法返回一个 FindIterable
实例,它提供了一个接口来链接和控制查找操作。使用 forEach()
方法可以提供一个 Block
作用于每个 document 并且迭代结束时执行回调一次。下面的代码遍历 collection 中所有的 document 并逐一打印,最后打印 “Operation Finished!”。
Block<Document> printDocumentBlock = new Block<Document>() {
@Override
public void apply(final Document document) {
System.out.println(document.toJson());
}
};
SingleResultCallback<Void> callbackWhenFinished = new SingleResultCallback<Void>() {
@Override
public void onResult(final Void result, final Throwable t) {
System.out.println("Operation Finished!");
}
};
collection.find().forEach(printDocumentBlock, callbackWhenFinished);
通过查询条件获得一个 document
我们可以创建一个过滤器传递给 find() 方法,以获得我们 collection 中的一组子集。例如,如果我们想查找 key为“i” ,value为71 的 document,我们要按下面的方法做(重用 printDocument
回调)。
import static com.mongodb.client.model.Filters.*;
collection.find(eq("i", 71)).first(printDocument);
最终会只印一个 document:
{ "_id" : { "$oid" : "5515836e58c7b4fbc756320b" }, "i" : 71 }
重要
请使用 Filters、Sorts、Projections 和 Updates API手册来找到简单、清晰的方法构建查询。
通过查询获得一组 documents
我们可以使用查询来从我们的 collection 中获得一组 document 集合。例如,如果我们想获得所有 key 为“i”,value 大于50 的 document ,我们应该按下面方式做(重用 printDocumentBlock
阻塞和 callbackWhenFinished
回调):
// 使用范围查询获取子集
collection.find(gt("i", 50)).forEach(printDocumentBlock, callbackWhenFinished);
范例应该会打印所有 i > 50
的document。
我们也可以增加上限范围,如 50 < i <= 100
:
collection.find(and(gt("i", 50), lte("i", 100))).forEach(printDocumentBlock, callbackWhenFinished);
document 排序
我们可以对 document 进行排序。通过在 FindIterable
上调用 sort()
方法,我们可以在一个查询上进行一次排序。
下面的例子中,我们使用 exists() 和 降序排序 descending("i") 来为我们的 document 排序。
collection.find(exists("i")).sort(descending("i")).first(printDocument);
投射域
有时我们不需要将所有的数据都存在一个 document 中。Projections 可以用来为查询操作构建投射参数并限制返回的字段。
下面的例子中,我们会对collection进行排序,排除 _id
字段,并输出第一个匹配的 document。
collection.find().projection(excludeId()).first(printDocument);
聚合
有时,我们需要将存储在 MongoDB 中的数据聚合。 Aggregates 支持对每种类型的聚合阶段进行构建。
下面的例子,我们执行一个两步骤的转换来计算 i * 10
的值。首先我们使用 Aggregates.match 查找所有 i > 0
的document 。接着,我们使用 Aggregates.project 结合 $multiply 操作来计算 “ITimes10
” 的值。
collection.aggregate(asList(
match(gt("i", 0)),
project(Document.parse("{ITimes10: {$multiply: ['$i', 10]}}")))
).forEach(printDocumentBlock, callbackWhenFinished);
For $group operations use the Accumulators helper for any accumulator operations.
对于 $group 操作使用 Accumulators 来处理任何 累加操作 。
下面的例子中,我们使用 Aggregates.group 结合 Accumulators.sum 来累加所有 i
的和。
collection.aggregate(singletonList(group(null, sum("total", "$i")))).first(printDocument);
注意
当前,还没有专门用于 聚合表达式 的工具类。可以使用 Document.parse() 来快速构建来自于JSON的聚合表达式。
更新 document
MongoDB 支持许多的 更新操作 。
要更新至多一个 document (可能没有匹配的document),使用 updateOne 方法指定过滤器并更新 document 。这里,我们使用 Updates.set 来更新匹配过滤器 i
等于 10 的第一个 document 并设置 i
的值为 110。
collection.updateOne(eq("i", 10), set("i", 110),
new SingleResultCallback<UpdateResult>() {
@Override
public void onResult(final UpdateResult result, final Throwable t) {
System.out.println(result.getModifiedCount());
}
});
使用 updateMany 方法可以更新所有匹配过滤器的 document 。这里我们使用 Updates.inc 来为所有 i
小于 100 的document 增加 100 。
collection.updateMany(lt("i", 100), inc("i", 100),
new SingleResultCallback<UpdateResult>() {
@Override
public void onResult(final UpdateResult result, final Throwable t) {
System.out.println(result.getModifiedCount());
}
});
更新方法返回一个 UpdateResult,其中包含了操作的信息(被修改的 document 的数量)。
删除 document
要删除至多一个 document (可能没有匹配的document)可以使用 deleteOne 方法。
collection.deleteOne(eq("i", 110), new SingleResultCallback<DeleteResult>() {
@Override
public void onResult(final DeleteResult result, final Throwable t) {
System.out.println(result.getDeletedCount());
}
});
使用 deleteMany 方法可以删除所有匹配过滤器的 document 。这里我们删除所有 i
大于等于的 document。
collection.deleteMany(gte("i", 100), new SingleResultCallback<DeleteResult>() {
@Override
public void onResult(final DeleteResult result, final Throwable t) {
System.out.println(result.getDeletedCount());
}
});
删除方法返回一个 DeleteResult,其中包含了操作的信息(被删除的 document 的数量)。
批量操作
批量操作允许批量的执行 插入、更新、删除操作。批量操作有两种类型:
- 有序的批量操作
有序的执行所有操作并在第一个写操作的错误处报告错误。
- 无序的批量操作
执行所有的操作并报告任何错误。
无序的批量操作不保证执行顺序。
我们来围观一下两个分别使用有序和无序操作的简单例子:
SingleResultCallback<BulkWriteResult> printBatchResult = new SingleResultCallback<BulkWriteResult>() {
@Override
public void onResult(final BulkWriteResult result, final Throwable t) {
System.out.println(result);
}
};
// 2. 有序批量操作
collection.bulkWrite(
Arrays.asList(new InsertOneModel<>(new Document("_id", 4)),
new InsertOneModel<>(new Document("_id", 5)),
new InsertOneModel<>(new Document("_id", 6)),
new UpdateOneModel<>(new Document("_id", 1),
new Document("$set", new Document("x", 2))),
new DeleteOneModel<>(new Document("_id", 2)),
new ReplaceOneModel<>(new Document("_id", 3),
new Document("_id", 3).append("x", 4))),
printBatchResult
);
// 2. 无序批量操作
collection.bulkWrite(
Arrays.asList(new InsertOneModel<>(new Document("_id", 4)),
new InsertOneModel<>(new Document("_id", 5)),
new InsertOneModel<>(new Document("_id", 6)),
new UpdateOneModel<>(new Document("_id", 1),
new Document("$set", new Document("x", 2))),
new DeleteOneModel<>(new Document("_id", 2)),
new ReplaceOneModel<>(new Document("_id", 3),
new Document("_id", 3).append("x", 4))),
new BulkWriteOptions().ordered(false),
printBatchResult
);
重要
不推荐在pre-2.6的MongoDB 服务器上使用 bulkWrite
方法。因为这是第一个支持批量写操作(插入、更新、删除)的服务器版本,它允许驱动去实现 BulkWriteResult
和 BulkWriteException
的语义。这个方法虽然仍然可以在pre-2.6服务器上工作,但是性能不好,一次只能执行一个写操作。
推荐阅读
-
MongoDB入门指南:快速上手
-
轻松上手Java单元测试:mock的快速指南
-
轻松上手!Java单元测试框架(Mockito)的快速指南
-
简单易学!Java后端Hello World应用的快速开发与部署指南
-
Java实现的QUIC快速协议指南
-
初学者快速上手指南:从零起步学Java
-
3个月快速掌握Java的学习路径图指南
-
如何轻松快速上手Java编程?初学者高效学习指南
-
【2022新手指南】Java编程进阶之路 - 六、技术架构篇 ### MySQL索引底层解析与优化实战 - 你会讲解MySQL索引的数据结构吗?性能调优技巧知多少? - Redis深度揭秘:你知道多少?从基础到哨兵、主从复制全梳理 - Redis持久化及哨兵模式详解,还有集群搭建和Leader选举黑箱打开 - Zookeeper是个啥?特性和应用场景大公开 - ZooKeeper集群搭建攻略及 Leader选举、读写一致性、共享锁实现细节 - 探究ZooKeeper中的Leader选举机制及其在分布式环境中的作用 - Zab协议深入剖析:原理、功能与在Zookeeper中的核心地位 - RabbitMQ全方位解读:工作模式、消费限流、可靠投递与配置策略 - 设计者视角:RabbitMQ过期时间、死信队列与延时队列实践指南 - RocketMQ特性和应用场景揭示:理解其精髓与差异化优势 - Kafka详细介绍:特性及广泛应用于实时数据处理的场景解析 - ElasticSearch实力揭秘:特性概述与作为搜索引擎的广泛应用 - MongoDB认知升级:非关系型数据库的优势阐述,安装与使用实战教学 - BIO/NIO/AIO网络模型对比:掌握它们的区别与在网络编程中的实际应用 - Netty带你飞:理解其超快速度背后的秘密,包括线程模型分析 - 网络通信黑科技:Netty编解码原理与常用编解码器的应用,Protostuff实战演示 - 解密Netty粘包与拆包现象,怎样有效应对这一常见问题 - 自定义Netty心跳检测机制,轻松调整检测间隔时间的艺术 - Dubbo轻骑兵介绍:核心特性概览,服务降级实战与其实现益处 - Dubbo三大神器解读:本地存根与本地伪装的实战运用与优势呈现 ----------------------- 七、结语与回顾
-
快速掌握JAVA基础:知识点全汇总指南