MongoDB Java API操作很全的整理
- 2019 年 11 月 4 日
- 筆記
MongoDB 是一个基于分布式文件存储的数据库。由 C++ 语言编写，一般生产上建议以共享分片的形式来部署。 但是MongoDB官方也提供了其它语言的客户端操作API。如下图所示：
提供了C、C++、C#、.net、GO、java、Node.js、PHP、python、scala等各种语言的版本，如下图所示：
MongoDB的操作分为同步操作和异步操作以及响应式编程操作
一、同步操作API
官方JAVA API的路径：https://docs.mongodb.com/ecosystem/drivers/java/ 我们这里以3.11的java 版本为例。各个版本的API对MongoDB服务的支持情况。
使用API时，先引入maven依赖
<!-- https://mvnrepository.com/artifact/org.mongodb/mongo-java-driver --> <dependency> <groupId>org.mongodb</groupId> <artifactId>mongo-java-driver</artifactId> <version>3.11.1</version> </dependency>
1、关于MongoDB Client的初始化和关闭。
从官方介绍来看，一般建议Client只需要一个建立一个长连接实例，然后使用时，都使用这个实例就可以，也就是可以用java的单例模式来创建连接实例。
//mongoClient连接 protected static MongoClient mongoClient; public synchronized static MongodbClient getInstance(String mongodbUrl) { if (null == mongoClient) { mongoClient = MongoClients.create(mongodbUrl); if(null != mongoClient){ log.info("mongoClient init success!"); } else{ log.info("mongoClient init failed!"); } } return mongodbClient; }
直接通过mongodb的host和port来创建client：
MongoClient mongoClient = MongoClients.create("mongodb://host1:27017");
client连接到一个 Replica Set：
MongoClient mongoClient = MongoClients.create("mongodb://host1:27017,host2:27017,host3:27017"); MongoClient mongoClient = MongoClients.create("mongodb://host1:27017,host2:27017,host3:27017/?replicaSet=myReplicaSet");
或者通过MongoClientSettings.builder() 来辅助生成连接字符串来创建client：
MongoClient mongoClient = MongoClients.create( MongoClientSettings.builder() .applyToClusterSettings(builder -> builder.hosts(Arrays.asList( new ServerAddress("host1", 27017), new ServerAddress("host2", 27017), new ServerAddress("host3", 27017)))) .build());
连接关闭：
public void close() { if(null!=mongoClient){ mongoClient.close(); mongoClient=null; } }
2、关于MongoDB 的基本操作
//创建Collection
public void createCollection(String dataBaseName,String collectionName){ getDatabase(dataBaseName).createCollection(collectionName); }
//查询dataBaseName
public MongoDatabase getDatabase(String dataBaseName){ return mongoClient.getDatabase(dataBaseName); }
//查询Collection
public List<String> listCollectionNames(String dataBaseName){
List<String> stringList = new ArrayList<String>();
mongoClient.getDatabase(dataBaseName).listCollectionNames().forEach((Consumer<? super String>) t->{ stringList.add(t); });
return stringList; }
public MongoCollection<Document> getCollectionByName(String dataBaseName, String collectionName){ return getDatabase(dataBaseName).getCollection(collectionName); }
3、关于MongoDB 的查询操作
//通过id（objectid）精确查询 public FindIterable<Document> findMongoDbDocById(String dataBaseName, String collectionName, String id){ BasicDBObject searchDoc = new BasicDBObject().append("_id", id); return getCollectionByName(dataBaseName,collectionName).find(searchDoc); } //通过id（objectid）模糊查询 public FindIterable<Document> findMongoDbDocByIdRegex(String dataBaseName, String collectionName, String id){ BasicDBObject searchDoc = new BasicDBObject().append("_id", new BasicDBObject("$regex",id)); return getCollectionByName(dataBaseName,collectionName).find(searchDoc); } //通过开始id和结束id 查询（根据objectId范围查询） public FindIterable<Document> findMongoDbDocById(String dataBaseName, String collectionName, String startId,String endId){ BasicDBObject searchDoc = new BasicDBObject().append("_id", new BasicDBObject("$gte", startId).append("$lte", endId)); return getCollectionByName(dataBaseName,collectionName).find(searchDoc); } public FindIterable<Document> findMongoDbDoc(String dataBaseName, String collectionName,BasicDBObject basicDBObject){ return getCollectionByName(dataBaseName,collectionName).find(basicDBObject); } //限制查询返回的条数 public FindIterable<Document> findMongoDbDoc(String dataBaseName, String collectionName,BasicDBObject basicDBObject,Integer limitNum){ return findMongoDbDoc(dataBaseName,collectionName,basicDBObject).limit(limitNum) ; } public FindIterable<Document> findMongoDbDocById(String dataBaseName, String collectionName, String startId,String endId,Integer limitNum){ return findMongoDbDocById(dataBaseName,collectionName,startId,endId).limit(limitNum); } /** * 降序查询（排序） * @param dataBaseName * @param collectionName * @param startId * @param endId * @param sortField 排序字段 * @return */ public FindIterable<Document> findMongoDbDocByIdDescSort(String dataBaseName, String collectionName, String startId,String endId,String sortField){ return findMongoDbDocById(dataBaseName,collectionName,startId,endId).sort(new Document().append(sortField, -1)); } public FindIterable<Document> findMongoDbDocByIdDescSort(String dataBaseName, String collectionName, String startId,String endId,String sortField,Integer limitNum){ return findMongoDbDocByIdDescSort(dataBaseName,collectionName,startId,endId,sortField).limit(limitNum); } /** * 降序查询（排序） * @param dataBaseName * @param collectionName * @param startId * @param endId * @param sortField 排序字段 * @return */ public FindIterable<Document> findMongoDbDocByIdAscSort(String dataBaseName, String collectionName, String startId,String endId,String sortField){ return findMongoDbDocById(dataBaseName,collectionName,startId,endId).sort(new Document().append(sortField, 1)); } public FindIterable<Document> findMongoDbDocByIdAscSort(String dataBaseName, String collectionName, String startId,String endId,String sortField,Integer limitNum){ return findMongoDbDocByIdAscSort(dataBaseName,collectionName,startId,endId,sortField).limit(limitNum); }
4、关于MongoDB 的插入操作
//插入操作，注意插入时，如果数据已经存在会报错，插入时必须数据不存在，不会自动进行覆盖 //插入单条记录 public void insertDoc(String dataBaseName, String collectionName, Document document){ getCollectionByName(dataBaseName,collectionName).insertOne(document); } //插入多条记录 public void insertDoc(String dataBaseName, String collectionName,List<? extends Document> listData){ getCollectionByName(dataBaseName,collectionName).insertMany(listData); }
5、关于MongoDB 的更新操作
//更新单条 public void updateDoc(String dataBaseName, String collectionName, Bson var1, Bson var2){ getCollectionByName(dataBaseName,collectionName).updateOne(var1,var2); } public void updateDoc(String dataBaseName, String collectionName, Bson var1, List<? extends Bson> list){ getCollectionByName(dataBaseName,collectionName).updateOne(var1,list); } //批量更新 public void updateDocs(String dataBaseName, String collectionName, Bson var1, Bson var2){ getCollectionByName(dataBaseName,collectionName).updateMany(var1,var2); } public void updateDocs(String dataBaseName, String collectionName, Bson var1, List<? extends Bson> list){ getCollectionByName(dataBaseName,collectionName).updateMany(var1,list); }
6、关于MongoDB 的删除操作
//单条删除 public DeleteResult deleteDoc(String dataBaseName, String collectionName, Bson var1){ return getCollectionByName(dataBaseName,collectionName).deleteOne(var1); } //批量删除 public DeleteResult deleteDocs(String dataBaseName, String collectionName,Bson var1){ return getCollectionByName(dataBaseName,collectionName).deleteMany(var1); }
7、关于MongoDB 的替换操作
//存在就替换，不存在的话就插入 public UpdateResult replaceDoc(String dataBaseName, String collectionName, Bson var1, Document var2){ return getCollectionByName(dataBaseName,collectionName).replaceOne(var1,var2); }
8、关于MongoDB 的bulkWrite操作 （批量写入）
public BulkWriteResult bulkWrite(String dataBaseName, String collectionName, List<? extends WriteModel<? extends Document>> listData){ return getCollectionByName(dataBaseName,collectionName).bulkWrite(listData); }
二、异步操作API
mongodb异步驱动程序提供了异步api，可以利用netty或java 7的asynchronoussocketchannel实现快速、无阻塞的i/o，maven依赖
<dependencies> <dependency> <groupId>org.mongodb</groupId> <artifactId>mongodb-driver-async</artifactId> <version>3.11.1</version> </dependency> </dependencies>
官方地址：http://mongodb.github.io/mongo-java-driver/3.11/driver-async/getting-started/installation/
异步操作必然会涉及到回调，回调时采用ResultCallback<Document>
SingleResultCallback<Document> callbackPrintDocuments = new SingleResultCallback<Document>() { @Override public void onResult(final Document document, final Throwable t) { 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!"); } };
异步insert操作
collection.insertMany(documents, new SingleResultCallback<Void>() { @Override public void onResult(final Void result, final Throwable t) { System.out.println("Documents inserted!"); } });
异步删除操作
collection.deleteMany(gte("i", 100), new SingleResultCallback<DeleteResult>() { @Override public void onResult(final DeleteResult result, final Throwable t) { System.out.println(result.getDeletedCount()); } });
异步更新操作
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()); } });
异步统计操作
collection.countDocuments( new SingleResultCallback<Long>() { @Override public void onResult(final Long count, final Throwable t) { System.out.println(count); } });
三、MongoDB Reactive Streams 操作API
官方的MongoDB reactive streams Java驱动程序，为MongoDB提供异步流处理和无阻塞处理。
完全实现reactive streams api，以提供与jvm生态系统中其他reactive streams的互操作，一般适合于大数据的处理，比如spark，flink，storm等。
<dependencies> <dependency> <groupId>org.mongodb</groupId> <artifactId>mongodb-driver-reactivestreams</artifactId> <version>1.12.0</version> </dependency> </dependencies>
官方地址：http://mongodb.github.io/mongo-java-driver-reactivestreams/
会包含如下三部分：
- Publisher：Publisher 是数据的发布者。Publisher 接口只有一个方法 subscribe，用于添加数据的订阅者，也就是 Subscriber。
- Subscriber： 是数据的订阅者。Subscriber 接口有4个方法，都是作为不同事件的处理器。在订阅者成功订阅到发布者之后，其 onSubscribe(Subscription s) 方法会被调用。
- Subscription：表示的是当前的订阅关系。
API问的地址：http://mongodb.github.io/mongo-java-driver-reactivestreams/1.12/javadoc/
代码示例：
//建立连接 MongoClient mongoClient = MongoClients.create(mongodbUrl); //获得数据库对象 MongoDatabase database = client.getDatabase(databaseName); //获得集合 MongoCollection collection = database.getCollection(collectionName); //异步返回Publisher FindPublisher publisher = collection.find(); //订阅实现 publisher.subscribe(new Subscriber() { @Override public void onSubscribe(Subscription str) { System.out.println("start..."); //执行请求 str.request(Integer.MAX_VALUE); } @Override public void onNext(Document document) { //获得文档 System.out.println("Document:" + document.toJson()); } @Override public void onError(Throwable t) { System.out.println("error occurs."); } @Override public void onComplete() { System.out.println("finished."); } });
