MongoDB Java Driver 源码分析(七)：com.mongodb.DBAPILayer

MongoDB Java Driver 源码分析(7)：com.mongodb.DBAPILayer
　　DBAPILayer 继承了抽象类 DB ，主要借助内部类 DBApiLayer.MyCollection 实现具体的数据库操作。
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 );        }

查看更多 下一篇