Vertx 接入MongoDB (九)

项目github地址：https://github.com/fotocj007/VertxWebApi

一：加入配置文件 mongodb.json

compile group: 'io.vertx', name: 'vertx-mongo-client', version: '3.9.8'

{
  "sources": [
    {"host": "10.0.205.199", "port": 27017, "db_name": "wsdemo","minPoolSize":5,"maxPoolSize": 10}
  ],
  "pools": 4
}

二：加载配置文件

public class MongoDbConfig extends JsonObjectConfig {
    public JsonArray sources;
    public int poolSize;

    public MongoDbConfig(Vertx vertx, String path){
        super(vertx,path);
    }

    @Override
    public void parse(JsonObject jsonObject) {
        sources = jsonObject.getJsonArray("sources");
        poolSize = jsonObject.getInteger("pools",8);
    }
}

三：链接mongodb

public class MongoPool {
    private int poolSize;

    private Vertx vertx;

    private List<JsonObject> sourceList;

    private Map<String,List<MongoClient>> pools;

    public List<String> dbList;

    public MongoPool(Vertx vertx, int poolS, List<JsonObject> sourceList){
        this.vertx = vertx;
        this.poolSize = poolS;
        this.sourceList = sourceList;
        pools = new HashMap<>(poolS);
        dbList = new ArrayList<>(2);
        initPool();
    }

    private void initPool() {
        for (JsonObject config : sourceList) {
            String db = config.getString("db_name");

            String connectionString = String.format("mongodb://%s:%d/%s",
                    config.getString("host"), config.getInteger("port"),
                    db);

            JsonObject options = new JsonObject()
                    .put("connection_string", connectionString)
                    .put("minPoolSize", config.getInteger("minPoolSize"))
                    .put("maxPoolSize",config.getInteger("maxPoolSize"));

            // look for username, password and auth_source
            Optional.ofNullable(config.getString("username", null))
                    .ifPresent(user -> options.put("username", user));
            Optional.ofNullable(config.getString("password", null))
                    .ifPresent(pass -> options.put("password", pass));

            List<MongoClient> list = new ArrayList<>();
            for (int j = 1; j <= poolSize; j++) {
                list.add(MongoClient.create(vertx, options));
            }

            dbList.add(db);
            pools.put(db, list);
        }
    }

    public MongoClient getClient(String db){
        return pools.get(db).get(ThreadLocalRandom.current().nextInt(poolSize));
    }

    public void close(){
        for(List<MongoClient> list : pools.values()){
            for(MongoClient client : list){
                client.close();
            }
        }
    }
}

四：添加个帮助类

public class PlayerMongo{
    protected static Logger logger = LoggerFactory.getLogger(PlayerMongo.class);

    private final String DB_NAME;

    protected MongoPool mongoDbPool;

    public PlayerMongo(MongoPool mongoPool) {
        this.DB_NAME = mongoPool.dbList.get(0);
        this.mongoDbPool = mongoPool;
    }

    public void findPlayerById(String collection,long playerId,Handler<AsyncResult<PlayerInfo>> handler) {
        JsonObject jsonQuery = new JsonObject().put("_id",playerId);
        mongoDbPool.getClient(DB_NAME).findOne(collection, jsonQuery, null, res -> {
            if (res.succeeded()) {
                JsonObject reData = res.result();

                long id = reData.getLong("_id",playerId);
                reData.remove("_id");

                PlayerInfo info = new JsonObject(reData.toString()).mapTo(PlayerInfo.class);
                info.setId(id);

                handler.handle(Future.succeededFuture(info));
            }else {
                handler.handle(Future.failedFuture(res.cause()));
                logger.error("findById error",res.cause());
            }
        });
    }

    public void updateAndInsert(String collection, long playerId, JsonObject upData, Handler<AsyncResult<Boolean>> handler){
        JsonObject find = new JsonObject().put("_id",playerId);
        JsonObject upInsert = new JsonObject()
                .put("$set",upData);

        //更新，没有时插入
        mongoDbPool.getClient(DB_NAME).findOneAndUpdateWithOptions(collection, find, upInsert,
                new FindOptions(),
                new UpdateOptions().setUpsert(true), res -> {
            if (res.succeeded()) {
                handler.handle(Future.succeededFuture(true));
            } else {
                handler.handle(Future.failedFuture(res.cause()));
                logger.error("updateAndInsert error",res.cause());
            }
        });
    }
}

五：Dao管理类

public class MongoManager {
    private MongoPool mongoPool;

    private PlayerMongo playerMongo;

    public MongoManager(MongoPool mySQLPool){
        this.mongoPool = mySQLPool;
        init();
    }

    private void init(){
        playerMongo = new PlayerMongo(mongoPool);
    }

    public PlayerMongo getPlayerMongo(){
        return playerMongo;
    }
}

六：修改configure，初始化mongodb.

public class Configure {
    private static final Configure ourInstance = new Configure();

    public Configure() {
    }

    public static Configure getInstance() {
        return ourInstance;
    }

    protected Vertx vertx;

    public MysqlConfig mysqlConfig;
    private MySQLUtil mySQLPool;
    public DaoManager daoManager;

    private RedisConfig redisConfig;
    private RedisPool redisPool;
    public RedisUtil redisUtil;

    private MongoDbConfig mongoDbConfig;
    private MongoPool mongoPool;
    public MongoManager mongoManager;

    public void init(Vertx vertx){
        this.vertx = vertx;

        initHandler();

        loadConfig();

        initDb();
        initRedis();
        initMongoDb();
    }

    private void initHandler(){
        HandlerManager.getInstance().addHandler(new DemoHandler());
    }

    /**
     *  加载db和Redis配置文件
     */
    protected void loadConfig(){
        mysqlConfig = new MysqlConfig(vertx, "res/mysql.json");
        redisConfig = new RedisConfig(vertx, "res/redis.json");
        mongoDbConfig = new MongoDbConfig(vertx,"res/mongodb.json");
    }

    protected void initDb(){
        List<JsonObject> list = new ArrayList<>();
        for(int i = 0; i< mysqlConfig.configs.size();i++){
            list.add(mysqlConfig.configs.getJsonObject(i));
        }
        mySQLPool = new MySQLUtil(vertx,2,list);

        daoManager = new DaoManager(mysqlConfig,mySQLPool);
    }

    /**
     *  初始化Redis
     */
    protected void initRedis(){
        redisPool = new RedisPool(vertx,redisConfig);
        redisUtil = new RedisUtil(redisPool);
    }

    private void initMongoDb(){
        List<JsonObject> list = new ArrayList<>();
        for(int i = 0; i< mongoDbConfig.sources.size();i++){
            list.add(mongoDbConfig.sources.getJsonObject(i));
        }
        mongoPool = new MongoPool(vertx,mongoDbConfig.poolSize,list);
        mongoManager = new MongoManager(mongoPool);
    }

    public void closeResource(){
        if(mySQLPool != null){
            mySQLPool.close();
        }
        if(redisPool != null){
            redisPool.close();
        }

        if(mongoPool != null){
            mongoPool.close();
        }
    }
}

七：测试一下，修改DemoHandler

        PlayerInfo info = new PlayerInfo();
        info.setId(3242353465L);
        info.setUserName("kkkkkdd");
        info.setAge(100);
        
        PlayerMongo playerMongo = Configure.getInstance().mongoManager.getPlayerMongo();
        playerMongo.updateAndInsert("playerColl",info.getId(),new JsonObject(JsonObject.mapFrom(info).toString()),res -> {
            System.out.println(res.result());


            //查询
            playerMongo.findPlayerById("playerColl",info.getId(),ress -> {
                System.out.println(ress);

            });
        });

发起请求，查看输出：

 

项目结构：