MongoDB Java Driver 源码分析(7):com.mongodb.DBAPILayer
程序员文章站
2022-03-02 16:13:25
...
DBAPILayer 继承了抽象类 DB ,主要借助内部类 DBApiLayer.MyCollection 实现具体的数据库操作。
DB 类的介绍可以参考:http://xxing22657-yahoo-com-cn.iteye.com/blog/1291183
DBApiLayer.MyCollection 继承了抽象类 DBCollection,具体实现了增删改查操作。
DBCollection 类的介绍可以参考之前的文章:http://xxing22657-yahoo-com-cn.iteye.com/blog/1255181
增删改查操作的方法声明如下:
这些操作都需要借助 DBTCPConnector 的 say 方法和 call 方法,这两个方法的实现可以参考这篇文章:
http://xxing22657-yahoo-com-cn.iteye.com/blog/1416331
下面以 insert 和 _find 操作为例进行分析
DBApiLayer.MyCollection._find 方法返回的是 Result 对象,它实现了 Iterator 接口,可以用于遍历。
next 方法和 hasNext 中都用到了 _advance 方法
DB 类的介绍可以参考:http://xxing22657-yahoo-com-cn.iteye.com/blog/1291183
借助 DBApiLayer.MyCollection 实现增删改查
DBApiLayer.MyCollection 继承了抽象类 DBCollection,具体实现了增删改查操作。
DBCollection 类的介绍可以参考之前的文章:http://xxing22657-yahoo-com-cn.iteye.com/blog/1255181
增删改查操作的方法声明如下:
// 插入操作 protected WriteResult insert(DBObject[] arr, boolean shouldApply , com.mongodb.WriteConcern concern ) // 删除操作 public WriteResult remove( DBObject o , com.mongodb.WriteConcern concern ) // 查找操作 Iterator<DBObject> __find( DBObject ref , DBObject fields , int numToSkip , int batchSize, int limit , int options ) // 更新操作 public WriteResult update( DBObject query , DBObject o , boolean upsert , boolean multi , com.mongodb.WriteConcern concern )
这些操作都需要借助 DBTCPConnector 的 say 方法和 call 方法,这两个方法的实现可以参考这篇文章:
http://xxing22657-yahoo-com-cn.iteye.com/blog/1416331
下面以 insert 和 _find 操作为例进行分析
// 插入操作 protected WriteResult insert(DBObject[] arr, boolean shouldApply , com.mongodb.WriteConcern concern ) throws MongoException { // 输出跟踪信息 if ( willTrace() ) { for (DBObject o : arr) { trace( "save: " + _fullNameSpace + " " + JSON.serialize( o ) ); } } // 是否更新对应的DBObject if ( shouldApply ){ for ( int i=0; i<arr.length; i++ ){ DBObject o=arr[i]; apply( o ); _checkObject( o , false , false ); Object id = o.get( "_id" ); if ( id instanceof ObjectId ){ ((ObjectId)id).notNew(); } } } WriteResult last = null; // 输出 DBObject 到 Mongo 服务器 int cur = 0; int maxsize = _mongo.getMaxBsonObjectSize(); while ( cur < arr.length ){ OutMessage om = new OutMessage( _mongo , 2002 ); // 以 0 作为交互开始的信号 om.writeInt( 0 ); // reserved // 输出完整的 namespace om.writeCString( _fullNameSpace ); // 将要输出的对象写入 OutMessage for ( ; cur<arr.length; cur++ ){ DBObject o = arr[cur]; om.putObject( o ); // 一次批量插入数据量的上限是 maxBsonObjectSize 的 4 倍 // 安全起见,这里使用 maxBsonObjectSize 的两倍 if ( om.size() > 2 * maxsize ){ // 超出一次批量插入的限制 // 停止构造 OutMessage,准备进入下一个循环 cur++; break; } } // 调用 DBTCPConnector 的 say 方法执行写入 last = _connector.say( _db , om , concern ); } return last; } // 查找操作 Iterator<DBObject> __find( DBObject ref , DBObject fields , int numToSkip , int batchSize, int limit , int options ) throws MongoException { if ( ref == null ) ref = new BasicDBObject(); // 输出跟踪信息 if ( willTrace() ) trace( "find: " + _fullNameSpace + " " + JSON.serialize( ref ) ); // 构造 OutMessage OutMessage query = OutMessage.query( _mongo , options , _fullNameSpace , numToSkip , chooseBatchSize(batchSize, limit, 0) , ref , fields ); // 调用 DBTCPConnector 的 call 方法获得查询结果 Response res = _connector.call( _db , this , query , null , 2 ); // 没有结果 if ( res.size() == 0 ) return null; // 检查错误 if ( res.size() == 1 ){ BSONObject foo = res.get(0); MongoException e = MongoException.parse( foo ); if ( e != null && ! _name.equals( "$cmd" ) ) throw e; } // 返回结果 return new Result( this , res , batchSize, limit , options ); }
借助 DBApiLayer.Result 遍历结果
DBApiLayer.MyCollection._find 方法返回的是 Result 对象,它实现了 Iterator 接口,可以用于遍历。
next 方法和 hasNext 中都用到了 _advance 方法
// 获取下一条记录 public DBObject next(){ // 当前 cursor 有下一条记录,直接返回 if ( _cur.hasNext() ) { return _cur.next(); } // 没有结果,抛异常 if ( ! _curResult.hasGetMore( _options ) ) throw new RuntimeException( "no more" ); // 有结果,但不在当前 cursor 中,取下一批数据 _advance(); // 递归调用 return next(); } // 是否包含下一条记录 public boolean hasNext(){ // 循环检查 while ( true ){ // 当前 cursor 有下一条记录,直接返回 true if ( _cur.hasNext() ) return true; // 没有结果,返回 false if ( ! _curResult.hasGetMore( _options ) ) return false; // 有结果,但不在当前 cursor 中,取下一批数据 _advance(); } } // 进行到下一条记录 private void _advance(){ if ( _curResult.cursor() <= 0 ) throw new RuntimeException( "can't advance a cursor <= 0" ); OutMessage m = new OutMessage( _mongo , 2005 ); // 以 0 作为交互开始的信号 m.writeInt( 0 ); // 输出完整的 namespace m.writeCString( _collection._fullNameSpace ); // 输出数据大小 m.writeInt( chooseBatchSize(_batchSize, _limit, _numFetched) ); // 输出当前 cusor 的位置 m.writeLong( _curResult.cursor() ); // 借助 DBTCPConnector 执行读取操作 Response res = _connector.call( DBApiLayer.this , _collection , m , _host ); // 读取下一条 _numGetMores++; // 初始化 init( res ); }
推荐阅读
-
互联网架构-Java8集合框架源码分析-041:纯手写Jdk7HashMap集合框架
-
centos7防火墙导致java程序访问mongodb3.0.1时报错的问题分析
-
MongoDB Java Driver 源码分析
-
MongoDB Java Driver 源码分析(9):com.mongodb.DBport
-
MongoDB Java Driver 源码分析(11):GridFS 类
-
MongoDB Java Driver 源码分析(13):OutputBuffer,BasicOutputBuffer 和 PoolOutputBuffer
-
MongoDB Java Driver 源码分析(1):Package 概述
-
MongoDB Java Driver 源码分析(12):GridFSFile、GridFSDBFile 和 GridFSInputFile
-
MongoDB Java Driver 源码分析(5):com.mongodb.DB
-
MongoDB Java Driver 源码分析(8):com.mongodb.RelicaSetStatus