mongodb-1

mongodb支持google提出的mapreduce并行编程，通过启动分片和水平扩展、读写分离，与关系数据库比较：table –> collection; row –>doucument/object(BSON)。

mongodb进程

mongod进程：负责整个mongodb中最核心的内容，负责数据库的创建，删除等，运行在服务器上进行监听。

mongod --config /home/honeycc/properties.conf
# properties.conf
# dbpath = /home/honeycc/db/data
# logpath = /home/honeycc/logs/log.log
# journal = true 启动数据库实例的日志功能，数据库当宕机后重启恢复
# port = 27017 
# auth = true  启动数据库实例的权限控制功能

mongo进程:交互式shell界面

1 2	mongo --port 27017 -u root -p root -autheticationDatabase gps_db

mongodump备份、mongoexport、mongoimport、mongos分片、mongofiles分布式文件存储系统接口、mongostat状态、mongotop性能跟踪

mongodump --port 27017 --db gps_db 
          --out /home/honeycc/back
mongoexport --port 27017 --db gps_db 
          --collection gps_data --out /home/honeycc/gps_data.json
mongoinport --port 27017 --db gps_db 
          --collection gps_data --file /home/honeycc/gps_data.json

mongodb 数据库操作

查询选择器

db.gps_data.find().pretty() 
# 格式化输出
db.gps_data.find({id:9,"province":"Zhejiang"}) 
# 精确查找
db.gps_data.find({"GPSTIME":{"&lt":new Date("2016-12-10")})
# 比较操作符"$gt" 、"$gte"、 "$lt"、 "$lte"、"$ne"(分别对应">"、 ">=" 、"<" 、"<="、"=")，组合起来进行范围的查找
db.gps_data.find({"GPSTIME":{"$in":[null],$exists:true}})
# 返回gpstime为null值的文档，如果$exists:false返回gpstime不为null的所有字段
#$in:[] ; $nin:[].单独使用$nin $ne会进行全表扫描,应与其他选择器配合使用 
db.gps_data.find({$and:[{},{}]})
# $or:[{},{}] $and:[{},{}]
db.gps_data.find("driver.message.age":20)
# 多层结构

查询投射

db.gps_data.find({"GPSTIME":{"&lt":new Date("2016-12-10")},{_id:0,GPSTIME:1,CODE:1,driver.message.phone:1,GPSTIME:{"$slice":-1}})
# db.gps_data.find({},{})
# 第一个{}内为查询选择器;第二个{}内为对前面返回的结构集进一步过滤条件，投射项,$slice:-1返回数组最后一条数据
db.gps_data.find({}).sort({id:-1})
# 排序效率低下，尽量确保排序的在索引上
db.gps_data.find({}).skip(10).limit(5).sort({id:-1})
# 先排序，然后跳过10条，获取5条，尽量少用skip，效率低

数组操作

db.gps_data.find({driver.message.phone:[110,112]})
# 数组精确匹配
db.gps_data.find({driver.message.phone:112})
# 数组包含匹配
    
db.gps_data.find({driver.message.phone.1:112})
# 数组位置匹配，第二个电话为112的

增删改

db.gps_data.insert({}) # _id 要唯一
 
db.gps_data.update(query,update,<upsert>,<multi>)
db.gps_data.update({CODE:"浙B26827"},{$set : { GPSTIME:new Date() },$inc:{driver.message.phone:100}}) 
#  第一个文档 更改车牌为浙B26827的，$set 把GPSTIME更改为当前时间，$inc 新增属性
db.gps_data.update({CODE:"浙B26827"},{CODE:"浙B26821"}) 
# 第一个文档 把其他属性都擦除掉了
 db.gps_data.update({CODE:"浙B26827"},{$set:{GPSTIME:new Date() },$inc:{driver.message.phone:100}},{upsert:true} ，{multi:true}) 
 # {multi:true} 所有文档 {upsert:true} 找不到则插入
db.gps_data.remove(<query>,<justOne>)
db.gps_data.remove({CODE:"浙B26827"},1)
# 删除全部、limt 1 删除一个

索引:提高数据获取的性能
在linux文件系统中，磁盘抽象为

引导块超级块索引节点表数据块

流程:索引节点表保存了所有文件或者目录对应的inode节点，通过文件名或目录找到对应的inode节点，通过inode节点定位到文件数据在文件系统中的逻辑块号，最后根据磁盘驱动程序将逻辑块号映射到磁盘具体的块号。

数据库上保存记录的机制是建立在文件系统上的，索引也是以文件的形式存储在磁盘上，运用最多的索引结构是B树。

Mongodb索引:B+树
db.gps_data.ensureIndex({})创建索引
db.gps_data.dropIndex(“GPSTIME_1_CODE_1 “)
1. 单字段索引 _id，唯一索引 {unique:true}: db.system.indexes.find()
2. 符合索引 db.gps_data.ensureIndex({GPSTIME:1,CODE:1})
3. 数组的多键索引对一个值为数组类型的字段创建索引，则默认会对数组中的每一个元素创建索引。

聚集分析:聚集操作是对数据进行分析的有效手段。mongodb主要提供了三种对数据进行分析的计算方式:管道模式聚集分析、Mapreduce聚集分析、简单函数和命令的聚集分析

管道模式:类unix上的管道命令 grep 层层过滤

         db.gps_data.aggregate([{$match:{}},{$group:{}}])
        '''
            常用管道操作符
            $match: 过滤文档，值传递匹配文档到管道中的下一步骤
            $limit: 限制管道中文档的数量
            $skip: 跳过指定数量文档
            $sort: 多所有文档进行排序
            $group: 对所有文档进行分组然后计算聚集结果
            $out: 将管道中的文档输出到一个具体集合中
            
            与$group操作一起使用的计算聚集值的操作符
            $first、$last、$max、$min、$avg、$sum
        '''
        
        # SQL语句与mongodb聚集操作语句比较
        #  select count(*) as count from gps_data
        db.gps_data.aggregate([
            {$group:{_id:null,count:{$sum:1}}}
            ])
        
        # select sum(num) as total from gps_data
        db.gps_data.aggregate([
            {$group:{_id:null,total:{$sum:"$num"}}}
            ])
        
        # select driver_id,sum(num) as total from gps_data group by driver_id
        db.gps_data.aggregate([
            {$group:{_id:"$driver_id",total:{$sum:"$num"}}}
        ])
        
         # select driver_id,gpsdate,sum(num) as total from gps_data group by driver_id
        db.gps_data.aggregate([
            {$group:{_id:{driver_id:"$driver_id",gpsdate:"gpsdate"},total:{$sum:"$num"}}}
            ])
        
         # select driver_id, count(*)   from gps_data group by driver_id having count(*) > 1
        db.gps_data.aggregate([
            {$group:{_id:"$driver_id",count:{$sum:1}}},
            {$match:{count :{$gt:1}}}            
            ])
```     
   
   2. Mapreduce模式聚集:用Mongodb做分布式存储，然后再用Mapreduce来做分析
```shell
        db.gps_data.mapReduce(
            # map 函数
            function(){
                emit(this.driver_id,this.num);
            },
            # reduce 函数
            funtion(key,values){
                return Array.sum(values)
            },
            {
                query:{ CODE:"浙B25681"}， # 查询条件
                outresult:"driver_car_total" # 输出结果到集合上
            }
        )
        <---->
        select sum(num) as value, driver_id as _id
        from gps_data
        where CODE="浙B25681"
        group by driver_id
        
        将结果输出到 driver_car_total集合上
        db.driver_cat_total.find()
        {
            "_id":1,"value":2
        }

简单聚集函数

# 1、distinct
# db.collection.distinct(key,<query>)
# 2、count
# db.collection.find(<query>).count()
# 3、group，结果集不能大于16M，不能在分片集群上进行操作且不能处理超过10000个唯一键值
# db.collection.group({key:..., initial:...,reduce:...[,cond:...]})
# eg:
db.gps_data.group({
    key:{_id:1},
    cond:{_id:{$lt:3}},
    reduce:function(cur,result){
        result.count += cur.count;
    }
    initial: {count:0}
})
# 统计_id小于3,按照_id分组求value值的和

JAVA操作Mongodb

Maven pom.xml配置驱动

  <dependency>
    <groupId>org.mongodb</groupId>
    <artifactId>mongo-java-driver</artifactId>
    <version>3.4.0</version>
</dependency>

Mongodb工具类

public class MongoDBUtil {
    private final static ThreadLocal<Mongo> mongos = new ThreadLocal<Mongo>();
    private static final String MONGODB_ADDRESS = "127.0.0.1";
    private static final int MONGODB_PORT = 27017;
    private static final String MONGODB_DBNAME = "gps_db";
public static DB getdb(){
    return getMongos().getDB(MONGODB_DBNAME);
}
public static Mongo getMongos() {
    Mongo mongo = mongos.get();
    if (mongo == null) {
        try {
            mongo = new Mongo(MONGODB_ADDRESS,MONGODB_PORT);
            mongos.set(mongo);
        } catch (MongoException e) {
            e.printStackTrace();
        }
    }
    return mongo;
}
public static void close(){
    Mongo mongo = mongos.get();
    if(mongo!=null){
        mongo.close();
        mongos.remove();
    }
}
/**
 * 获取集合（表）
 *
 * @param collection
 */
public static DBCollection getCollection(String collection) {
    return getdb().getCollection(collection);
}
/**
 * 插入
 *
 * @param collection
 * @param o 插入
 *
 */
public static void insert(String collection, DBObject o) {
    getCollection(collection).insert(o);
}
/**
 * 批量插入
 *
 * @param collection
 * @param list
 *            插入的列表
 */
public void insertBatch(String collection, List<DBObject> list) {
    if (list == null || list.isEmpty()) {
        return;
    }
    getCollection(collection).insert(list);
}
/**
 * 删除
 *
 * @param collection
 * @param q
 *            查询条件
 */
public void delete(String collection, DBObject q) {
    getCollection(collection).remove(q);
}
/**
 * 批量删除
 *
 * @param collection
 * @param list
 *            删除条件列表
 */
public void deleteBatch(String collection, List<DBObject> list) {
    if (list == null || list.isEmpty()) {
        return;
    }
    for (int i = 0; i < list.size(); i++) {
        getCollection(collection).remove(list.get(i));
    }
}
/**
 * 更新
 *
 * @param collection
 * @param q
 *            查询条件
 * @param setFields
 *            更新对象
 */
public static void update(String collection, DBObject q, DBObject setFields) {
    getCollection(collection).updateMulti(q,
            new BasicDBObject("$set", setFields));
}
/**
 * 查找集合所有对象
 *
 * @param collection
 */
public static List<DBObject> findAll(String collection) {
    return getCollection(collection).find().toArray();
}
/**
 * 按顺序查找集合所有对象
 *
 * @param collection
 *            数据集
 * @param orderBy
 *            排序
 */
public static List<DBObject> findAll(String collection, DBObject orderBy) {
    return getCollection(collection).find().sort(orderBy)
            .toArray();
}
/**
 * 查找（返回一个对象）
 *
 * @param collection
 * @param q
 *            查询条件
 */
public static DBObject findOne(String collection, DBObject q) {
    return getCollection(collection).findOne(q);
}
/**
 * 查找（返回一个对象）
 *
 * @param collection
 * @param q
 *            查询条件
 * @param fileds
 *            返回字段
 */
public static DBObject findOne(String collection, DBObject q, DBObject fileds) {
    return getCollection(collection).findOne(q, fileds);
}
/**
 * 分页查找集合对象，返回特定字段
 *
 * @param collection
 * @param q
 *            查询条件
 * @param fileds
 *            返回字段
 * @pageNo 第n页
 * @perPageCount 每页记录数
 */
public static List<DBObject> findLess(String collection, DBObject q, DBObject fileds, int pageNo,
                                      int perPageCount) {
    return getCollection(collection).find(q, fileds)
            .skip((pageNo - 1) * perPageCount).limit(perPageCount)
            .toArray();
}
public static long getCollectionCount(String collection) {
    return getCollection(collection).getCount();
}

使用mongodb插件 mongodb plugin

  
<dependency>
    <groupId>org.mongodb</groupId>
    <artifactId>mongo-java-driver</artifactId>
    <version>3.4.0</version>
</dependency>
<dependency>
    <groupId>org.slf4j</groupId>
    <artifactId>slf4j-api</artifactId>
    <version>1.7.21</version>
</dependency>
<dependency>
    <groupId>com.cybermkd</groupId>
    <artifactId>MongodbPlugin</artifactId>
    <version>1.0.8.0</version>
</dependency>

       //初始化
    public static MongoClient init(){
        MongoPlugin mongoPlugin=new MongoPlugin();
        mongoPlugin.add(MONGODB_ADDRESS,MONGODB_PORT);
        mongoPlugin.setDatabase(MONGODB_DBNAME);
        MongoClient client = mongoPlugin.getMongoClient();
        MongoKit.INSTANCE.init(client, mongoPlugin.getDatabase());
        return  client;
    }
    //使用
    public JSONArray getCollecitonGroupBy() {
        JSONArray jsonArray = new JSONArray();
        MongoClient client = null;
        try {
            client = MongoDBUtil.init();
            MongoQuery query = new MongoQuery();
            query.use("test").ascending("GPSTIME").find();
            MongoAggregation aggregation=new MongoAggregation(query);
            MongoQuery queryGroupBy = aggregation
                    .group(new Document("CODE","$CODE").append("GPSDATE","$GPSDATE"))
                    .getQuery();
            List<JSONObject> jsonList  = queryGroupBy
                    .projection("GPSDATE","CODE","GPSTIME","country","city","province","district","LNG","LAT")
                    .findAll();
            System.out.println(jsonList.get(0));
            jsonArray.add(jsonList.get(0));
            return jsonArray;
        }finally {
            client.close();
        }
    }
    
    
    
public static void main( String args[] ){
        try{
            /*
              // 连接到 mongodb 服务
            MongoClient mongoClient = new MongoClient( "localhost" , 27017 );
            // 连接到数据库
            MongoDatabase mongoDatabase = mongoClient.getDatabase("test");
            MongoCollection<Document> collection = mongoDatabase.getCollection("test");
            System.out.println(collection.count());
            BasicDBObject key =  new BasicDBObject("CODE",true);
   
              FindIterable<Document> findIterable = collection.find();
            MongoCursor<Document> mongoCursor = findIterable.iterator();
            while(mongoCursor.hasNext()){
                System.out.println(mongoCursor.next());
            }
            DBObject dbObject = new BasicDBObject("province","浙江省");
            System.out.print(MongoDBUtil.findOne("test",dbObject).toString());
*/
          
            MongoPlugin mongoPlugin=new MongoPlugin();
            mongoPlugin.add("127.0.0.1",27017);
            mongoPlugin.setDatabase("test");
            MongoClient client = mongoPlugin.getMongoClient();
            MongoKit.INSTANCE.init(client, mongoPlugin.getDatabase());
            MongoQuery query = new MongoQuery();
            query.use("test")./*eq("CODE","Y115浙B92236").*/ascending("GPSTIME").find();
            MongoAggregation aggregation=new MongoAggregation(query);
            System.out.print(JSON.toJSONString(aggregation
                    .include("GPSDATE","CODE","GPSTIME","country","city","province","district","LNG","LAT")
                    .projection()
                    //.pipeline(new Document("$match",new Document("provice", "浙江省")))
                    .pipeline(new Document("$group",
                            new Document("_id",new Document("province", "$province")
                                    .append("GPSDATE","$GPSDATE"))
                                    .append("total",new Document("$sum",1))))
                    .out("output")
                    .aggregate()));
           /* MongoQuery queryGroupBy = aggregation
                    .group(new Document("CODE","$CODE").append("GPSDATE","$GPSDATE"))
                    .getQuery();
            List<JSONObject> jsonList  = queryGroupBy
                    .projection("GPSDATE","CODE","GPSTIME","country","city","province","district","LNG","LAT")
                    .findAll();
            System.out.println(jsonList.get(0));*/
            client.close();
        }catch(Exception e){
            System.err.println( e.getClass().getName() + ": " + e.getMessage() );
        }
    }

mongodb api文档教程
 mongodb plugin 插件使用教程

总结

第一次接触nosql数据库，还有很多需要去学习东西，在java上去开发还不是那么流畅，插件的使用还不是很顺畅，但是运用起来还是感觉很舒服。
在数据量打的情况下，只做数据处理的前提，最好是先在数据库去将数据处理好，再用java mongodb api 去调用数据库，获取处理好的数据，避免的在逻辑层去处理大量的数据，这样的性能效率才能高