[譯] 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 的數量）。

批量操作

批量操作允許批量的執行 插入、更新、刪除操作。批量操作有兩種類型：

1. 有序的批量操作

   有序的執行所有操作並在第一個寫操作的錯誤處報告錯誤。

2. 無序的批量操作

   執行所有的操作並報告任何錯誤。

   無序的批量操作不保證執行順序。

我們來圍觀一下兩個分別使用有序和無序操作的簡單例子：

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服務器上工作，但是性能不好，一次只能執行一個寫操作。