java操作mongodb基本实现。 <!-- more -->

package com.demo.utils.db;

import com.mongodb.Block;
import com.mongodb.MongoClient;
import com.mongodb.MongoClientURI;
import com.mongodb.ReadPreference;
import com.mongodb.bulk.BulkWriteResult;
import com.mongodb.client.*;
import com.mongodb.client.model.*;
import org.apache.commons.lang3.StringUtils;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.bson.Document;
import org.bson.conversions.Bson;
import org.bson.types.ObjectId;
import org.joda.time.DateTime;

import java.util.ArrayList;
import java.util.List;

public class MongoHelper {

    private Logger logger = LogManager.getLogger(this);

    public final static String AUTO_ID = "_id";

    public final static String UpdateDate = "_UpdateDate";

    public final static int MaxSize = 1000;

    private String hosts;

    private String databaseName;

    private MongoClient masterClusterClient;
    private MongoDatabase masterClusterDB;

    private MongoClient readClusterClient;
    private MongoDatabase readClusterDB;

    public MongoHelper(String hosts, String databaseName) {

        this.hosts = hosts;
        this.databaseName = databaseName;

        masterClusterClient = new MongoClient(new MongoClientURI(this.hosts));
        masterClusterClient.setReadPreference(ReadPreference.secondaryPreferred());
        masterClusterDB = masterClusterClient.getDatabase(this.databaseName);
        this.readClusterClient = this.masterClusterClient;
        this.readClusterDB = this.masterClusterDB;
    }

    public MongoHelper(String masterClusterHosts, String masterClusterDatabaseName, String readClusterHosts, String readClusterDatabaseName) {
        if (StringUtils.isEmpty(masterClusterHosts)) {
            throw new IllegalArgumentException("the 'masterClusterHosts' is null or empty");
        }
        if (StringUtils.isEmpty(masterClusterDatabaseName)) {
            throw new IllegalArgumentException("the 'masterClusterDatabaseName' is null or empty");
        }
        if (logger.isDebugEnabled()) {
            logger.debug("masterClusterHosts:{}", masterClusterHosts);
            logger.debug("masterClusterDatabaseName:{}", masterClusterDatabaseName);
        }
        masterClusterClient = new MongoClient(new MongoClientURI(masterClusterHosts));
        masterClusterClient.setReadPreference(ReadPreference.secondaryPreferred());
        masterClusterDB = masterClusterClient.getDatabase(masterClusterDatabaseName);
        if (StringUtils.isEmpty(readClusterHosts) || StringUtils.isEmpty(readClusterDatabaseName)) {
            this.readClusterClient = this.masterClusterClient;
            this.readClusterDB = this.masterClusterDB;
        } else {
            if (logger.isDebugEnabled()) {
                logger.debug("readClusterHosts:{}", readClusterHosts);
                logger.debug("readClusterDatabaseName:{}", readClusterDatabaseName);
            }
            if (masterClusterHosts.equals(readClusterHosts) && masterClusterDatabaseName.equals(readClusterDatabaseName)) {
                this.readClusterClient = this.masterClusterClient;
                this.readClusterDB = this.masterClusterDB;
            } else {
                readClusterClient = new MongoClient(new MongoClientURI(readClusterHosts));
                readClusterDB = readClusterClient.getDatabase(readClusterDatabaseName);
            }
        }
    }

    /**
    * 执行mongo命令
    *
    * @param document
    * @return
    */
    public Document runCommand(Document document) {
        if (document == null || document.isEmpty()) {
            throw new IllegalArgumentException("the 'document' can not be null or empty");
        }
        if (document.containsKey("insert") || document.containsKey("update") || document.containsKey("delete")) {
            return this.masterClusterDB.runCommand(document, ReadPreference.primary());
        } else {
            return this.readClusterDB.runCommand(document, ReadPreference.secondaryPreferred());
        }
    }

    /**
    * 获取数据库名称列表
    *
    * @return 所有数据库名称列表 需要权限
    */
    public List<String> getAllDBNames() {
        MongoIterable<String> dbnames = masterClusterClient.listDatabaseNames();
        if (dbnames == null) {
            return new ArrayList<String>();
        }
        List<String> _list = new ArrayList<String>();
        for (String s : dbnames) {
            _list.add(s);
        }
        return _list;
    }

    /**
    * 获取mongo集群DB实例
    *
    * @return
    */
    public MongoDatabase getDB() {
        return getDB(databaseName, false);
    }

    /**
    * 获取mongo集群DB实例
    *
    * @param dbName 数据库名大小写
    * @param isRead 是否读库
    * @return
    */
    public MongoDatabase getDB(String dbName, Boolean isRead) {
        if (dbName != null && !"".equals(dbName)) {
            MongoDatabase database = isRead ? masterClusterClient.getDatabase(dbName) : masterClusterClient.getDatabase(dbName);
            return database;
        }
        return null;
    }

    /**
    * @return 查询默认DB下的所有集合
    */
    public List<String> getAllCollectionNames() {
        MongoIterable<String> colls = masterClusterDB.listCollectionNames();
        if (colls == null) {
            return new ArrayList<String>();
        }
        List<String> _list = new ArrayList<String>();
        for (String s : colls) {
            _list.add(s);
        }
        return _list;
    }

    /**
    * 查询指定DB下的所有集合
    *
    * @param dbName 指定DB
    * @return 指定DB下的所有集合名称
    */
    public List<String> getAllCollectionNames(String dbName) {
        MongoDatabase db = masterClusterClient.getDatabase(dbName);
        if (db == null) {
            return new ArrayList<String>();
        }
        MongoIterable<String> colls = db.listCollectionNames();
        if (colls == null) {
            return new ArrayList<String>();
        }
        List<String> _list = new ArrayList<String>();
        for (String s : colls) {
            _list.add(s);
        }
        return _list;
    }

    /**
    * 从默认读库获取collection集合对象
    *
    * @param collName 集合名字大小写
    * @return 集合对象
    */
    private MongoCollection<Document> getCollection(String collName) {
        return getCollection(collName, ReadPreference.secondaryPreferred());
    }

    /**
    * 获取Mongodb集合
    *
    * @param collName      集合名称
    * @param readPreference 读操作优先顺序
    * @return
    */
    private MongoCollection<Document> getCollection(String collName, ReadPreference readPreference) {
        if (StringUtils.isEmpty(collName)) {
            throw new IllegalArgumentException("the 'collName' is null or empty");
        }
        MongoCollection<Document> collection = this.masterClusterDB.getCollection(collName).withReadPreference(readPreference);
        return collection;
    }

    /**
    * 获取用于读的集合。readClusterDB
    *
    * @param collName
    * @return
    */
    private MongoCollection<Document> getCollectionOfRead(String collName) {
        if (StringUtils.isEmpty(collName)) {
            throw new IllegalArgumentException("the 'collName' is null or empty");
        }
        MongoCollection<Document> collection = this.readClusterDB.getCollection(collName);
        return collection;
    }

    /**
    * 获取用于写的集合。ReadPreference.primary
    *
    * @param collName
    * @return
    */
    private MongoCollection<Document> getCollectionOfWrite(String collName) {
        if (StringUtils.isEmpty(collName)) {
            throw new IllegalArgumentException("the 'collName' is null or empty");
        }
        MongoCollection<Document> collection = this.masterClusterDB.getCollection(collName).withReadPreference(ReadPreference.primary());
        return collection;
    }

    /**
    * 从默认读库获取collection集合对象
    *
    * @param collName 集合名字大小写
    * @return 集合对象
    */
    public List<Document> getDocuments(String collName, Document condition) {
        final List<Document> docs = new ArrayList<>();
        if (StringUtils.isEmpty(collName)) {
            return docs;
        }

        FindIterable<Document> iterable = this.getCollectionOfRead(collName).find(condition);
        iterable.forEach(new Block<Document>() {

            @Override
            public void apply(final Document doc) {
                docs.add(doc);
            }
        });
        return docs;
    }

    /**
    * 向指定集合中插入一条数据
    *
    * @param collName 集合名称
    * @param doc      要插入的数据
    * @return 是否插入成功
    */
    public String insertOne(String collName, Document doc) {
        MongoCollection<Document> collection = this.getCollectionOfWrite(collName);
        if (collection == null) {
            return null;
        }
        doc.append(UpdateDate, DateTime.now().toString("yyyy-MM-dd HH:mm:ss.SSS"));
        collection.insertOne(doc);
        if (doc.containsKey(MongoHelper.AUTO_ID)) {
            return doc.get(MongoHelper.AUTO_ID).toString();
        }
        return null;
    }

    /**
    * 向指定集合中插入多条数据
    *
    * @param collName 集合名称
    * @param docs    要插入的数据集合
    * @return 插入是否成功
    */
    public boolean insertMany(String collName, List<Document> docs) {
        MongoCollection<Document> collection = this.getCollectionOfWrite(collName);
        if (collection == null) {
            return false;
        }
        String dateStr = DateTime.now().toString("yyyy-MM-dd HH:mm:ss.SSS");
        for (Document doc : docs) {
            doc.append(UpdateDate, dateStr);
        }
        collection.insertMany(docs);
        return true;
    }

    /**
    * 查询集合的第一个文档
    *
    * @param collName 集合名称
    * @return 返回第一个文档
    */
    public Document findOne(String collName) {
        MongoCollection<Document> coll = this.getCollectionOfRead(collName);
        if (coll == null) {
            return null;
        }
        FindIterable<Document> iterable = coll.find().limit(1);
        if (iterable == null) {
            return null;
        }
        return iterable.first();
    }

    /**
    * 查询文档
    *
    * @param collName
    * @param filter
    * @return
    */
    public Document findOne(String collName, Bson filter) {
        MongoCollection<Document> coll = this.getCollectionOfRead(collName);
        if (coll == null) {
            return null;
        }
        FindIterable<Document> iterable = coll.find(filter).limit(1);
        if (iterable == null) {
            return null;
        }
        return iterable.first();
    }

    /**
    * 查询文档
    *
    * @param collName 集合名称
    * @param filter  过滤条件
    * @param fields  输出字段
    * @return
    */
    public Document findOne(String collName, Bson filter, String... fields) {
        MongoCollection<Document> coll = this.getCollectionOfRead(collName);
        if (coll == null) {
            return null;
        }
        FindIterable<Document> iterable = coll.find(filter).limit(1);
        if (iterable == null) {
            return null;
        }
        return iterable.projection(Projections.include(fields)).first();
    }

    /**
    * 查询文档
    *
    * @param collName  集合名称
    * @param filter    过滤条件
    * @param projection 字段投影
    * @return
    */
    public Document findOne(String collName, Bson filter, Bson projection) {
        MongoCollection<Document> coll = this.getCollectionOfRead(collName);
        if (coll == null) {
            return null;
        }
        FindIterable<Document> iterable = coll.find(filter).limit(1);
        if (iterable == null) {
            return null;
        }
        return iterable.projection(projection).first();
    }

    /**
    * 查询文档
    *
    * @param collName    集合名称
    * @param filter      过滤条件
    * @param projections 字段投影
    * @param orderField  排序字段
    * @param ascding    是否升序
    * @return
    */
    public Document findOne(String collName, Bson filter, Bson projections, String orderField, boolean ascding) {
        MongoCollection<Document> coll = this.getCollectionOfRead(collName);
        if (coll == null) {
            return null;
        }
        FindIterable<Document> iterable = coll
                .find(filter)
                .sort(ascding ? Sorts.ascending(orderField) : Sorts.descending(orderField))
                .limit(1);
        if (iterable == null) {
            return null;
        }
        return iterable.projection(projections).first();
    }

    /**
    * 查询指定id的文档
    *
    * @param collName 集合名称
    * @param id      文档主键id
    * @return 
    */
    public Document findRowById(String collName, String id) {
        if (ObjectId.isValid(id)) {
            MongoCollection<Document> coll = this.getCollectionOfRead(collName);
            if (coll == null) {
                return null;
            }
            FindIterable<Document> iterable = coll.find(Filters.eq("_id", new ObjectId(id))).limit(1);
            if (iterable == null) {
                return null;
            }
            return iterable.first();
        } else {
            return null;
        }
    }

    /**
    * 查询文档集合
    *
    * @param collName    集合名称
    * @param filter      过滤条件
    * @param projections 字段投影
    * @return 最多返回 @MaxSize 个文档，避免内存占用过大。返回结果按照_id字段降序排列
    */
    public List<Document> findRows(String collName, Bson filter, Bson projections) {
        return findRows(collName, filter, projections, null, false);
    }

    /**
    * 查询文档集合，并按照指定字段排序
    *
    * @param collName    集合名称
    * @param filter      过滤条件
    * @param projections 字段投影
    * @param orderField  排序字段
    * @param ascding    是否升序
    * @return 最多返回  @MaxSize 个文档，避免内存占用过大
    */
    public List<Document> findRows(String collName, Bson filter, Bson projections, String orderField, boolean ascding) {
        List<Document> list = new ArrayList<Document>();
        MongoCollection<Document> coll = this.getCollectionOfRead(collName);
        if (coll == null) {
            return list;
        }
        FindIterable<Document> iterable = null;
        if (filter == null) {
            iterable = coll.find();
        } else {
            iterable = coll.find(filter);
        }
        if (iterable == null) {
            return list;
        }
        MongoCursor<Document> cursor = null;
        try {
            if (StringUtils.isEmpty(orderField)) {
                if (projections == null) {
                    cursor = iterable
                            .sort(ascding ? Sorts.ascending("_id") : Sorts.descending("_id"))
                            .limit(MaxSize)
                            .iterator();
                } else {
                    cursor = iterable.projection(projections)
                            .sort(ascding ? Sorts.ascending("_id") : Sorts.descending("_id"))
                            .limit(MaxSize)
                            .iterator();
                }
            } else {
                if (projections == null) {
                    cursor = iterable
                            .sort(ascding ? Sorts.ascending(orderField) : Sorts.descending(orderField))
                            .limit(MaxSize)
                            .iterator();
                } else {
                    cursor = iterable.projection(projections)
                            .sort(ascding ? Sorts.ascending(orderField) : Sorts.descending(orderField))
                            .limit(MaxSize)
                            .iterator();
                }
            }
            if (cursor != null && cursor.hasNext()) {
                while (cursor.hasNext()) {
                    list.add(cursor.next());
                }
            }
            return list;
        } finally {
            if (cursor != null)
                cursor.close();
        }
    }

    /**
    * @param collName 集合名称
    * @param filter  过滤条件
    * @return
    */
    public long getRowsCount(String collName, Bson filter) {
        MongoCollection<Document> coll = this.getCollectionOfRead(collName);
        if (coll == null) {
            return 0;
        }
        if (filter == null) {
            return coll.count();
        }
        return coll.count(filter);
    }

    /**
    * 查询分页文档集合,按照_id字段降序排列
    *
    * @param collName    集合名称
    * @param filter      过滤条件
    * @param projections 字段投影
    * @param pageNo      当前页编号，从0开始
    * @param pageSize    每页容量
    * @return
    * @throws Exception
    */
    public List<Document> findPagedRows(String collName, Bson filter, Bson projections, int pageNo, int pageSize) {
        return findPagedRows(collName, filter, projections, null, false, pageNo, pageSize);
    }

    /**
    * 查询分页文档集合,默认按照_id字段排序
    *
    * @param collName    集合名称
    * @param filter      过滤条件
    * @param projections 字段投影
    * @param orderField  排序字段
    * @param ascding    是否升序
    * @param pageNo      当前页编号，从0开始
    * @param pageSize    每页容量
    * @return
    */
    public List<Document> findPagedRows(String collName, Bson filter, Bson projections, String orderField, boolean ascding, int pageNo, int pageSize) {
        List<Document> list = new ArrayList<Document>(pageSize);
        MongoCollection<Document> coll = this.getCollectionOfRead(collName);
        if (coll == null) {
            return list;
        }
        FindIterable<Document> iterable = null;
        if (filter == null) {
            iterable = coll.find();
        } else {
            iterable = coll.find(filter);
        }
        if (iterable == null) {
            return list;
        }
        MongoCursor<Document> cursor = null;
        try {
            if (StringUtils.isEmpty(orderField)) {
                cursor = iterable.projection(projections)
                        .sort(ascding ? Sorts.ascending("_id") : Sorts.descending("_id"))
                        .skip((pageNo - 1) * pageSize)
                        .limit(pageSize)
                        .iterator();
            } else {
                cursor = iterable.projection(projections)
                        .sort(Sorts.ascending(orderField))
                        .sort(ascding ? Sorts.ascending(orderField) : Sorts.descending(orderField))
                        .skip((pageNo - 1) * pageSize)
                        .limit(pageSize)
                        .iterator();
            }
            if (cursor != null && cursor.hasNext()) {
                while (cursor.hasNext()) {
                    list.add(cursor.next());
                }
            }
            return list;
        } finally {
            if (cursor != null)
                cursor.close();
        }
    }

    /**
    * 根据主键id删除文档
    *
    * @param collName 集合名称
    * @param id      文档id
    * @return 被删除的文档数量
    */
    public long deleteById(String collName, String id) {
        if (ObjectId.isValid(id)) {
            MongoCollection<Document> coll = this.getCollectionOfWrite(collName);
            if (coll == null) {
                return 0;
            }
            return coll.deleteOne(Filters.eq("_id", new ObjectId(id)))
                    .getDeletedCount();
        } else {
            return 0;
        }
    }

    /**
    * 根据指定filter删除文档
    *
    * @param collName
    * @param filter
    * @return 被删除文档的数量
    */
    public long delete(String collName, Bson filter) {
        if (filter == null) {
            return 0;
        } else {
            MongoCollection<Document> coll = this.getCollectionOfWrite(collName);
            if (coll == null) {
                return 0;
            }
            return coll.deleteOne(filter)
                    .getDeletedCount();
        }
    }

    /**
    * 更新文档
    *
    * @param collName 集合名称
    * @param filter  过滤条件
    * @param update  更新语句
    * @return 更新的文档数量
    */
    public long update(String collName, Bson filter, Bson update) {
        MongoCollection<Document> coll = this.getCollectionOfWrite(collName);
        if (coll == null) {
            return 0;
        }
        return coll.updateOne(filter, Updates.combine(
                update,
                Updates.set(UpdateDate, DateTime.now().toString("yyyy-MM-dd HH:mm:ss.SSS")))
        ).getModifiedCount();
    }

    /**
    * 更新文档
    *
    * @param collName 集合名称
    * @param updates  更新语句
    * @return 更新的文档数量
    */
    public long updateMany(String collName, List<WriteModel<Document>> updates) {
        MongoCollection<Document> coll = this.getCollectionOfWrite(collName);
        if (coll == null || updates == null || updates.size() == 0) {
            return 0;
        }
        BulkWriteResult bulkWriteResult = coll.bulkWrite(updates);
        return bulkWriteResult.getModifiedCount();
    }

    /**
    * 更新文档，指定文档存在则更新，不存在则插入
    *
    * @param collName 集合名称
    * @param filter  过滤条件
    * @param update  更新语句
    * @return 更新的文档数量
    */
    public long upsert(String collName, Bson filter, Bson update) {
        MongoCollection<Document> coll = this.getCollectionOfWrite(collName);
        if (coll == null) {
            return 0;
        }
        ;
        return coll.updateOne(filter, Updates.combine(
                update,
                Updates.set(UpdateDate, DateTime.now().toString("yyyy-MM-dd HH:mm:ss.SSS"))
                ),
                new UpdateOptions().upsert(true)
        ).getModifiedCount();
    }

    public List<Document> aggregate(String collName, List<Bson> pipelines) {
        MongoCollection<Document> coll = this.getCollectionOfRead(collName);
        MongoCursor<Document> cursor = null;
        List<Document> documents = new ArrayList<>();
        try {
            cursor = coll.aggregate(pipelines).iterator();
            if (cursor != null && cursor.hasNext()) {
                while (cursor.hasNext()) {
                    documents.add(cursor.next());
                }
            }
            return documents;
        } finally {
            if (cursor != null)
                cursor.close();
        }
    }
}