MongoDB是一种文本式数据库。与传统的关系式数据库最大不同是MongoDB没有标准的格式要求，即没有schema，合适高效处理当今由互联网+商业产生的多元多态数据。MongoDB也是一种分布式数据库，充分具备大数据处理能力和高可用性。MongoDB提供了scala终端驱动mongo-scala-driver，我们就介绍一下MongoDB数据库和通过scala来进行数据操作编程。
与关系数据库相似，MongoDB结构为Database->Collection->Document。Collection对应Table，Document对应Row。因为MongoDB没有schema，所以Collection中的Document可以是不同形状格式的。在用scala使用MongoDB之前必须先建立连接，scala-driver提供了多种连接方式：
val client1 = MongoClient() val client2 = MongoClient("mongodb://localhost:27017") val clusterSettings = ClusterSettings.builder() .hosts(List(new ServerAddress("localhost:27017")).asJava).build() val clientSettings = MongoClientSettings.builder().clusterSettings(clusterSettings).build() val client = MongoClient(clientSettings)
下面是一些对应的MongoClient构建函数：
/** * Create a default MongoClient at localhost:27017 * * @return MongoClient */ def apply(): MongoClient = apply("mongodb://localhost:27017") /** * Create a MongoClient instance from a connection string uri * * @param uri the connection string * @return MongoClient */ def apply(uri: String): MongoClient = MongoClient(uri, None) /** * Create a MongoClient instance from a connection string uri * * @param uri the connection string * @param mongoDriverInformation any driver information to associate with the MongoClient * @return MongoClient * @note the `mongoDriverInformation` is intended for driver and library authors to associate extra driver metadata with the connections. */ def apply(uri: String, mongoDriverInformation: Option[MongoDriverInformation]): MongoClient = {...} /** * Create a MongoClient instance from the MongoClientSettings * * @param clientSettings MongoClientSettings to use for the MongoClient * @return MongoClient */ def apply(clientSettings: MongoClientSettings): MongoClient = MongoClient(clientSettings, None) /** * Create a MongoClient instance from the MongoClientSettings * * @param clientSettings MongoClientSettings to use for the MongoClient * @param mongoDriverInformation any driver information to associate with the MongoClient * @return MongoClient * @note the `mongoDriverInformation` is intended for driver and library authors to associate extra driver metadata with the connections. */ def apply(clientSettings: MongoClientSettings, mongoDriverInformation: Option[MongoDriverInformation]): MongoClient = {
与MongoDB建立连接后可用选定Database：
val db = client.getDatabase("testdb")
由于没有格式限制，所以testdb不需要预先构建，像文件系统的directory一样，不存在时可以自动创建。同样，db内的collection也是可以自动创建的，因为不需要预先设定字段格式（no-schema）：
val db: MongoDatabase = client.getDatabase("testdb") val userCollection: MongoCollection[Document] = db.getCollection("users")
collection中Document类的构建函数如下：
/** * Create a new document from the elems * @param elems the key/value pairs that make up the Document. This can be any valid `(String, BsonValue)` pair that can be * transformed into a [[BsonElement]] via [[BsonMagnets.CanBeBsonElement]] implicits and any [[BsonTransformer]]s that * are in scope. * @return a new Document consisting key/value pairs given by `elems`. */ def apply(elems: CanBeBsonElement*): Document = { val underlying = new BsonDocument() elems.foreach(elem => underlying.put(elem.key, elem.value)) new Document(underlying) }
Document可以通过CanbeBsonElement构建。CanbeBsonElement是一种key/value结构：
/** * Represents a single [[BsonElement]] * * This is essentially a `(String, BsonValue)` key value pair. Any pair of `(String, T)` where type `T` has a [[BsonTransformer]] in * scope into a [[BsonValue]] is also a valid pair. */
sealed trait CanBeBsonElement { val bsonElement: BsonElement /** * The key of the [[BsonElement]] * @return the key */ def key: String = bsonElement.getName /** * The value of the [[BsonElement]] * @return the BsonValue */ def value: BsonValue = bsonElement.getValue } /** * Implicitly converts key/value tuple of type (String, T) into a `CanBeBsonElement` * * @param kv the key value pair * @param transformer the implicit [[BsonTransformer]] for the value * @tparam T the type of the value * @return a CanBeBsonElement representing the key/value pair */
implicit def tupleToCanBeBsonElement[T](kv: (String, T))(implicit transformer: BsonTransformer[T]): CanBeBsonElement = { new CanBeBsonElement { override val bsonElement: BsonElement = BsonElement(kv._1, transformer(kv._2)) } }
有了上面这个tupleToCanBeBsonElement隐式转换函数就可以用下面的方式构建Document了：
val doc: Document = Document("_id" -> 0, "name" -> "MongoDB", "type" -> "database", "count" -> 1, "info" -> Document("x" -> 203, "y" -> 102))
这种key/value关系对应了一般数据库表中的字段名称/字段值。下面我们尝试建两个不同格式的Document并把它们加入到同一个collection里：
val alice = Document("_id" -> 1, "name" -> "alice wong", "age" -> 24) val tiger = Document("first" -> "tiger", "last" -> "chan", "name" -> "tiger chan", "age" -> "unavailable") val addAlice: Observable[Completed] = userCollection.insertOne(alice) val addTiger: Observable[Completed] = userCollection.insertOne(tiger)
上面这个例子证明了MongoDB的no-schema特性。用insert方法加入数据返回结果是个Obervable类型。这个类型与Future很像：只是一种运算的描述，必须通过subscribe方法来实际运算获取结果：
addAlice.subscribe(new Observer[Completed] { override def onComplete(): Unit = println("insert alice completed.") override def onNext(result: Completed): Unit = println("insert alice sucessful.") override def onError(e: Throwable): Unit = println(s"insert error: ${e.getMessage}") })
又或者转成Future后用Future方法如Await来运算：
def headResult(observable: Observable[Completed]) = Await.result(observable.head(), 2 seconds) val r1 = headResult(addTiger)
Mongo-Scala提供了Observable到Future的转换函数：
/** * Collects the [[Observable]] results and converts to a [[scala.concurrent.Future]]. * * Automatically subscribes to the `Observable` and uses the [[collect]] method to aggregate the results. * * @note If the Observable is large then this will consume lots of memory! * If the underlying Observable is infinite this Observable will never complete. * @return a future representation of the whole Observable */ def toFuture(): Future[Seq[T]] = { val promise = Promise[Seq[T]]() collect().subscribe((l: Seq[T]) => promise.success(l), (t: Throwable) => promise.failure(t)) promise.future } /** * Returns the head of the [[Observable]] in a [[scala.concurrent.Future]]. * * @return the head result of the [[Observable]]. */ def head(): Future[T] = { import scala.concurrent.ExecutionContext.Implicits.global headOption().map { case Some(result) => result case None => null.asInstanceOf[T] // scalastyle:ignore null
} }
也可以用insertMany来成批加入：
val peter = Document("_id" -> 3, "first" -> "peter", "age" -> "old") val chan = Document("last" -> "chan", "family" -> "chan's") val addMany = userCollection.insertMany(List(peter,chan)) val r2 = headResult(addMany)
现在我们可以用count得出usersCollection中Document数量和用find把所有Document都印出来：
userCollection.count.head.onComplete { case Success(c) => println(s"$c documents in users collection") case Failure(e) => println(s"count() error: ${e.getMessage}") } userCollection.find().toFuture().onComplete { case Success(users) => users.foreach(println) case Failure(e) => println(s"find error: ${e.getMessage}") } scala.io.StdIn.readLine()
显示结果：
insert alice sucessful. insert alice completed. 4 documents in users collection Document((_id,BsonInt32{value=1}), (name,BsonString{value='alice wong'}), (age,BsonInt32{value=24})) Document((_id,BsonObjectId{value=5a96641aa83f2923ab437602}), (first,BsonString{value='tiger'}), (last,BsonString{value='chan'}), (name,BsonString{value='tiger chan'}), (age,BsonString{value='unavailable'})) Document((_id,BsonInt32{value=3}), (first,BsonString{value='peter'}), (age,BsonString{value='old'})) Document((_id,BsonObjectId{value=5a96641aa83f2923ab437603}), (last,BsonString{value='chan'}), (family,BsonString{value='chan's'}))
这个BsonString很碍眼，用隐式转换来把它转成String：
object Helpers { implicit class DocumentObservable[C](val observable: Observable[Document]) extends ImplicitObservable[Document] { override val converter: (Document) => String = (doc) => doc.toJson } implicit class GenericObservable[C](val observable: Observable[C]) extends ImplicitObservable[C] { override val converter: (C) => String = (doc) => doc.toString } trait ImplicitObservable[C] { val observable: Observable[C] val converter: (C) => String def results(): Seq[C] = Await.result(observable.toFuture(), 10 seconds) def headResult() = Await.result(observable.head(), 10 seconds) def printResults(initial: String = ""): Unit = { if (initial.length > 0) print(initial) results().foreach(res => println(converter(res))) } def printHeadResult(initial: String = ""): Unit = println(s"${initial}${converter(headResult())}") } }
现在再列印：
userCollection.find().printResults("all documents:") all documents:{ "_id" : 1, "name" : "alice wong", "age" : 24 } { "_id" : { "$oid" : "5a9665cea83f29243ccacbd2" }, "first" : "tiger", "last" : "chan", "name" : "tiger chan", "age" : "unavailable" } { "_id" : 3, "first" : "peter", "age" : "old" } { "_id" : { "$oid" : "5a9665cea83f29243ccacbd3" }, "last" : "chan", "family" : "chan's" }
现在可读性强多了。find()无条件选出所有Document。MongoDB-Scala通过Filters对象提供了完整的查询条件构建函数如equal：
/** * Creates a filter that matches all documents where the value of the field name equals the specified value. Note that this does * actually generate a `\$eq` operator, as the query language doesn't require it. * * A friendly alias for the `eq` method. * * @param fieldName the field name * @param value the value * @tparam TItem the value type * @return the filter * @see [[http://docs.mongodb.org/manual/reference/operator/query/eq \$eq]] */ def equal[TItem](fieldName: String, value: TItem): Bson = eq(fieldName, value)
equal返回Bson，我们也可以把多个Bson组合起来形成一个更复杂的查询条件：
userCollection.find(and(gte("age",24),exists("name",true)))
好了，现在我们可以测试各种查询条件了：
userCollection.find(notEqual("_id",3)).printResults("id != 3:") userCollection.find(equal("last", "chan")).printResults("last = chan:") userCollection.find(and(gte("age",24),exists("name",true))).printResults("age >= 24") userCollection.find(or(gte("age",24),equal("first","tiger"))).printResults("first = tiger")
显示结果：
id != 3:{ "_id" : 1, "name" : "alice wong", "age" : 24 } { "_id" : { "$oid" : "5a9665cea83f29243ccacbd2" }, "first" : "tiger", "last" : "chan", "name" : "tiger chan", "age" : "unavailable" } { "_id" : { "$oid" : "5a9665cea83f29243ccacbd3" }, "last" : "chan", "family" : "chan's" } last = chan:{ "_id" : { "$oid" : "5a9665cea83f29243ccacbd2" }, "first" : "tiger", "last" : "chan", "name" : "tiger chan", "age" : "unavailable" } { "_id" : { "$oid" : "5a9665cea83f29243ccacbd3" }, "last" : "chan", "family" : "chan's" } age >= 24{ "_id" : 1, "name" : "alice wong", "age" : 24 } first = tiger{ "_id" : 1, "name" : "alice wong", "age" : 24 } { "_id" : { "$oid" : "5a9665cea83f29243ccacbd2" }, "first" : "tiger", "last" : "chan", "name" : "tiger chan", "age" : "unavailable" }
下面是本次示范的源代码：
build.sbt
name := "learn-mongo" version := "0.1" scalaVersion := "2.12.4" libraryDependencies := Seq( "org.mongodb.scala" %% "mongo-scala-driver" % "2.2.1", "com.lightbend.akka" %% "akka-stream-alpakka-mongodb" % "0.17" )
MongoScala101.scala
import org.mongodb.scala._ import scala.collection.JavaConverters._ import org.mongodb.scala.connection.ClusterSettings import scala.concurrent._ import scala.concurrent.duration._ import scala.util._ import org.mongodb.scala.model.Filters._ object MongoScala101 extends App { import scala.concurrent.ExecutionContext.Implicits.global
// val client1 = MongoClient() // val client2 = MongoClient("mongodb://localhost:27017")
val clusterSettings = ClusterSettings.builder() .hosts(List(new ServerAddress("localhost:27017")).asJava).build() val clientSettings = MongoClientSettings.builder().clusterSettings(clusterSettings).build() val client = MongoClient(clientSettings) val db: MongoDatabase = client.getDatabase("testdb") val userCollection: MongoCollection[Document] = db.getCollection("users") val deleteAll = userCollection.deleteMany(notEqual("_id", 3)) deleteAll.head.onComplete { case Success(c) => println(s"delete sucessful $c") case Failure(e) => println(s"delete error: ${e.getMessage}") } scala.io.StdIn.readLine() val delete3 = userCollection.deleteMany(equal("_id", 3)) delete3.head.onComplete { case Success(c) => println(s"delete sucessful $c") case Failure(e) => println(s"delete error: ${e.getMessage}") } scala.io.StdIn.readLine() val doc: Document = Document("_id" -> 0, "name" -> "MongoDB", "type" -> "database", "count" -> 1, "info" -> Document("x" -> 203, "y" -> 102)) val alice = Document("_id" -> 1, "name" -> "alice wong", "age" -> 24) val tiger = Document("first" -> "tiger", "last" -> "chan", "name" -> "tiger chan", "age" -> "unavailable") val addAlice: Observable[Completed] = userCollection.insertOne(alice) val addTiger: Observable[Completed] = userCollection.insertOne(tiger) addAlice.subscribe(new Observer[Completed] { override def onComplete(): Unit = println("insert alice completed.") override def onNext(result: Completed): Unit = println("insert alice sucessful.") override def onError(e: Throwable): Unit = println(s"insert error: ${e.getMessage}") }) def headResult(observable: Observable[Completed]) = Await.result(observable.head(), 2 seconds) val r1 = headResult(addTiger) val peter = Document("_id" -> 3, "first" -> "peter", "age" -> "old") val chan = Document("last" -> "chan", "family" -> "chan's") val addMany = userCollection.insertMany(List(peter,chan)) val r2 = headResult(addMany) import Helpers._ userCollection.count.head.onComplete { case Success(c) => println(s"$c documents in users collection") case Failure(e) => println(s"count() error: ${e.getMessage}") } userCollection.find().toFuture().onComplete { case Success(users) => users.foreach(println) case Failure(e) => println(s"find error: ${e.getMessage}") } scala.io.StdIn.readLine() userCollection.find().printResults("all documents:") userCollection.find(notEqual("_id",3)).printResults("id != 3:") userCollection.find(equal("last", "chan")).printResults("last = chan:") userCollection.find(and(gte("age",24),exists("name",true))).printResults("age >= 24") userCollection.find(or(gte("age",24),equal("first","tiger"))).printResults("first = tiger") client.close() println("end!!!") } object Helpers { implicit class DocumentObservable[C](val observable: Observable[Document]) extends ImplicitObservable[Document] { override val converter: (Document) => String = (doc) => doc.toJson } implicit class GenericObservable[C](val observable: Observable[C]) extends ImplicitObservable[C] { override val converter: (C) => String = (doc) => doc.toString } trait ImplicitObservable[C] { val observable: Observable[C] val converter: (C) => String def results(): Seq[C] = Await.result(observable.toFuture(), 10 seconds) def headResult() = Await.result(observable.head(), 10 seconds) def printResults(initial: String = ""): Unit = { if (initial.length > 0) print(initial) results().foreach(res => println(converter(res))) } def printHeadResult(initial: String = ""): Unit = println(s"${initial}${converter(headResult())}") } }