Vertx 接入Redis (八)

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

web服务器经典的mysql+redis模式，这里介绍redis的接入。

一：导入gradle

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

二：添加配置文件

{
  "sources": [
    {"connectionString": "redis://:foto_CJ123@10.0.201.215:6379/0", "maxPoolSize":6},
    {"connectionString": "redis://:foto_CJ123@10.0.201.215:6379/1", "maxPoolSize":6},
    {"connectionString": "redis://:foto_CJ123@10.0.201.215:6379/2", "maxPoolSize":6},
    {"connectionString": "redis://:foto_CJ123@10.0.201.215:6379/3", "maxPoolSize":6}
  ],
  "poolSize": 8
}

三：添加json文件加载 RedisConfig

public class RedisConfig extends JsonObjectConfig {
    private ArrayList<RedisOptions> sources;

    private int poolSize;

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

    @Override
    public void parse(JsonObject jsonObject) {
        poolSize = jsonObject.getInteger("poolSize");
        sources = new ArrayList<>();
        JsonArray sourcesJa = jsonObject.getJsonArray("sources");
        for (int i = 0; i < sourcesJa.size();i++) {
            JsonObject jo = sourcesJa.getJsonObject(i);

            RedisOptions redisOpt = new RedisOptions(jo);

            sources.add(redisOpt);
        }
    }

    public ArrayList<RedisOptions> getSources() {
        return sources;
    }

    public int getPoolSize() {
        return poolSize;
    }
}

四：添加RedisPool

public class ConsistentHashWithNode {
    private static final String VIR_NODE_NAME_SEPARATOR = "@VirNode";

    private static final List<String> instanceInfo = new LinkedList<>();

    /**
     * 初始化虚拟节点 key表示服务器虚拟节点的hash值，value表示服务器虚拟节点的名称
     */
    private static final SortedMap<Integer, String> serverHashMap = new TreeMap<>();
    /**
     * 设置每台服务器需要的虚拟节点
     */
    private static final int VIRTUAL_NODES = 1000;

    /**
     * 构建hash环
     * @param servers
     */
    public ConsistentHashWithNode(List<String> servers) {
        //首先本地缓存一份实例信息
        instanceInfo.addAll(servers);
        instanceInfo.forEach(instance -> {
            for (int i = 0; i < VIRTUAL_NODES; i++) {
                //构建虚拟节点
                String virNodeName = instance + VIR_NODE_NAME_SEPARATOR + i;
                serverHashMap.put(hash(virNodeName), virNodeName);
            }
        });
    }

    public static void addNode(String server) {
        //首先本地缓存一份实例信息
        instanceInfo.add(server);
        for (int i = 0; i < VIRTUAL_NODES; i++) {
            //构建虚拟节点
            String virNodeName = server + VIR_NODE_NAME_SEPARATOR + i;
            serverHashMap.put(hash(virNodeName), virNodeName);
        }
    }

    /**
     * 根据数据获取真实存储服务节点
     * @param data
     * @return
     */
    public static String getServer(String data) {
        Integer firstKey;
        SortedMap<Integer, String> subSortedMap = serverHashMap.tailMap(hash(data));
        if (subSortedMap.isEmpty()){
            firstKey = serverHashMap.firstKey();
        }else{
            firstKey = subSortedMap.firstKey();
        }

        String virNodeName = serverHashMap.get(firstKey);
        return virNodeName.substring(0, virNodeName.indexOf(VIR_NODE_NAME_SEPARATOR));
    }

    /**
     * FNV1_32_HASH 百度
     * @param str
     * @return
     */
    public static int hash(String str) {
        final int p = 16777619;
        int hash = (int)2166136261L;
        for (int i = 0; i < str.length(); i++) {
            hash = (hash ^ str.charAt(i)) * p;
        }
        hash += hash << 13;
        hash ^= hash >> 7;
        hash += hash << 3;
        hash ^= hash >> 17;
        hash += hash << 5;
        if (hash < 0)
            hash = Math.abs(hash);
        return Math.abs(hash);
    }
}

public class RedisPool {
    private int poolSize;

    private final Vertx vertx;

    private final List<RedisOptions> dataSources;

    private final Map<String,List<RedisAPI>> pools;

    /*************************
     * Redis连接池参数配置和初始化
     */
    public RedisPool(Vertx vertx, RedisConfig redisConfig){
        this.vertx = vertx;
        this.poolSize = redisConfig.getPoolSize();
        this.dataSources = redisConfig.getSources();
        pools = new HashMap<>();
        initPool();
    }

    private void initPool(){
        for (RedisOptions dataSource : dataSources) {
            String server = dataSource.getEndpoint();

            List<RedisAPI> list = new ArrayList<>();
            for (int j = 1; j <= poolSize; j++) {
                RedisClient client = new RedisClient(vertx,dataSource);

                list.add(RedisAPI.api(client));
            }
            pools.put(server, list);
            ConsistentHashWithNode.addNode(server);
        }
    }

    /*************************
     * 根据下标获取也给链接
     */
    public RedisAPI getClientByIndex(int dbId){
        RedisOptions options = dataSources.get(dbId);
        String server = options.getEndpoint();
        return pools.get(server).get(ThreadLocalRandom.current().nextInt(poolSize));
    }

    /*************************
     * 根据hashKey获取一个链接
     */
    public RedisAPI getClient(String key){
        String server = ConsistentHashWithNode.getServer(key);
        return pools.get(server).get(ThreadLocalRandom.current().nextInt(poolSize));
    }

    /*************************
     * 关闭连接
     */
    public void close(){
        for(List<RedisAPI> list : pools.values()){
            for(RedisAPI client : list){
                client.close();
            }
        }
    }
}

五：Redis帮助类

public class RedisUtil {
    private final Logger logger = LoggerFactory.getLogger(RedisUtil.class);

    private final RedisPool redisPool;

    public RedisUtil(RedisPool redisPool){
        this.redisPool = redisPool;
    }

    /**
     * db = 0
     * 数据格式 key-value
     */
    public void setConfigValue(String rKey, String value,long ex){
        redisPool.getClientByIndex(0).setex(rKey,String.valueOf(ex),value,res -> {
            if(!res.succeeded()) {
                logger.error("setConfigValue key="+rKey,res.cause());
            }
        });
    }

    public void getConfigValue(String rKey, Handler<AsyncResult<Response>> handler){
        redisPool.getClientByIndex(0).get(rKey,res -> {
            if(res.succeeded()) {
                handler.handle(Future.succeededFuture(res.result()));
            }else {
                handler.handle(Future.failedFuture(res.cause()));
                logger.error("getConfigValue key="+rKey,res.cause());
            }
        });
    }

    /**
     * 分库
     * 数据格式 key-value
     */
    public void setValueStrById(String passportId,String rKey, String value,long ex){
        redisPool.getClient(passportId).setex(rKey,String.valueOf(ex),value,res -> {
            if(!res.succeeded()) {
                logger.error("setPlayerValue key="+rKey,res.cause());
            }
        });
    }

    public void getValueStrById(String passportId,String rKey,Handler<AsyncResult<Response>> handler){
        redisPool.getClient(passportId).get(rKey,res -> {
            if(res.succeeded()) {
                handler.handle(Future.succeededFuture(res.result()));
            }else {
                handler.handle(Future.failedFuture(res.cause()));
                logger.error("getValueStrById key="+rKey,res.cause());
            }
        });
    }

}

六：初始化，修改Config

public class Configure {
    private static final Configure ourInstance = new 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;

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

        initHandler();

        loadConfig();

        initDb();
        initRedis();
    }

    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");
    }

    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);
    }
}

七：测试一下，修改DemoHandler

    PlayerInfo info = new PlayerInfo();
        info.setUserName("kkkkkdd");
        info.setAge(100);

//        PlayerDao client = Configure.getInstance().daoManager.getPlayerDao();
//        client.saveBaseEntity(info,res -> {
//
//        });

        String key = "demo_test_key";
        Configure.getInstance().redisUtil.setConfigValue(key, JsonObject.mapFrom(info).toString(),300);

        try {
            Thread.sleep(1000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

        Configure.getInstance().redisUtil.getConfigValue(key,res -> {
            System.out.println(res.result().toString());

            PlayerInfo rInfo = new JsonObject(res.result().toString()).mapTo(PlayerInfo.class);
            System.out.println(rInfo.getUserName());
        });

发起请求：

 

项目结构：