public class MongoDB001 {
private Mongo mongo;
/**
* 连接mongodb
* @param host
* @param port
* @throws UnknownHostException
*/
public MongoDB001(String host, int port) throws UnknownHostException {
mongo = new MongoClient(host, port);
}
/**
* 获取DB
* @param dbname
* @return
*/
public DB getDB(String dbname) {
return mongo.getDB(dbname);
}
/**
* 获取collection,如果不存在，则创建
* @param db
* @param collectionName
* @return
*/
public DBCollection getCollection(DB db, String collectionName) {
if(!db.collectionExists(collectionName)){
return db.createCollection(collectionName, new BasicDBObject("capped", false).append("size", 1048576));
}
return db.getCollection(collectionName);
}
/**
* 插入数据
* @param coll
* @return
*/
public int insert(DBCollection coll) {
BasicDBObject doc = new BasicDBObject("name", "MongoDB")
.append("age", 25).append("province", "上海").append("home", "江西").append("_id", 1);
WriteResult result = coll.insert(doc);
return result.getN();
}
/**
* 更新数据
* @param coll
* @param id
*/
public void update(DBCollection coll, int id){
DBObject o = coll.findOne(new BasicDBObject("_id", id));
coll.update(new BasicDBObject("_id", id), new BasicDBObject("province", "湖南").append("name", o.get("name"))
.append("age", o.get("age")).append("home", o.get("home")));
}
/**
* 批量插入
* @param coll
* @return
*/
public int bulkInsert(DBCollection coll){
BulkWriteOperation builder = coll.initializeOrderedBulkOperation();
BasicDBObject object = null;
for(int i = 1;i<100; i++){
object = new BasicDBObject("_id",i);
builder.insert(object.append("name", "测试"+i).append("age", i).append("province", "上海").append("home", "上海"));
}
BulkWriteResult result = builder.execute();
return result.getInsertedCount();
}
/**
* 游标查询
* @param coll
*/
public void query(DBCollection coll){
DBCursor cursor = coll.find(new BasicDBObject("id", 2).append("type", "database"));
try{
while (cursor.hasNext()) {
System.out.println(cursor.next().get("_id"));
}
}finally{
cursor.close();
}
}
/**
* 删除DB
* @param dbName
*/
public void dropDatabase(String dbName){
mongo.dropDatabase(dbName);
}
/**
* 删除collection
* @param coll
*/
public void dropCollection(DBCollection coll){
coll.drop();
}
/**
* 根据id，删除数据
* @param coll
* @param id
* @return
*/
public int delete(DBCollection coll, int id){
return coll.remove(new BasicDBObject().append("_id", id)).getN();
}
public void parallelScanOptions(DBCollection coll){
ParallelScanOptions options = ParallelScanOptions.builder().batchSize(2).numCursors(2).build();
List<Cursor> list = coll.parallelScan(options);
for(Cursor cursor : list){
while(cursor.hasNext()){
System.out.println(cursor.next());
}
}
}
}
