demo-tabledata-redis/src/main/java/com/fr/plugin/db/redis/core/RedisPool.java

package com.fr.plugin.db.redis.core;

import com.fanruan.api.conf.ConfigurationKit;
import com.fanruan.api.generic.Matcher;
import com.fanruan.api.generic.Runner;
import com.fanruan.api.log.LogKit;
import com.fanruan.api.util.ArrayKit;
import com.fanruan.api.util.StringKit;
import com.fr.config.Configuration;
import com.fr.plugin.db.redis.core.accessor.category.ClusterRedisClient;
import com.fr.plugin.db.redis.core.accessor.category.StandaloneRedisClient;
import com.fr.plugin.db.redis.core.emb.Redis;
import com.fr.plugin.db.redis.core.emb.impl.ProxyRedis;
import com.fr.plugin.db.redis.core.emb.impl.SimpleRedis;
import com.fr.plugin.db.redis.core.pool.RedisConnectionPoolConfig;
import com.fr.plugin.db.redis.core.pool.RedisConnectionProxyConfig;
import com.fr.stable.collections.combination.Pair;
import com.jcraft.jsch.JSch;
import com.jcraft.jsch.JSchException;
import com.jcraft.jsch.Session;
import org.apache.commons.pool2.impl.GenericObjectPoolConfig;
import redis.clients.jedis.HostAndPort;
import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisCluster;
import redis.clients.jedis.JedisPool;
import redis.clients.jedis.JedisPoolConfig;

import java.util.ArrayList;
import java.util.LinkedHashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;

/**
 * @author richie
 * @version 10.0
 * Created by richie on 2019-03-18
 */
public class RedisPool {

    private static RedisPool pool = new RedisPool();

    static {
        ConfigurationKit.listenCacheChange(new Matcher<Class<? extends Configuration>>() {
            @Override
            public boolean match(Class clazz) {
                return RedisConnectionPoolConfig.class == clazz;
            }
        }, new Runner() {
            @Override
            public void run() {
                pool.clearAll();
            }
        });
    }

    private Map<String, JedisPool> jedisPoolMap = new ConcurrentHashMap<String, JedisPool>();

    private void clearAll() {
        jedisPoolMap.clear();
    }

    public static Redis getFinal(boolean isCluster, String host, String port, String password) {
        List<Pair<String, Integer>> pairs = findAllHostAndPassword(host, port);
        if (isCluster) {
            return pool.getJedisCluster(pairs, password);
        } else if (pairs.size() == 1) {
            return pool.getStandaloneResource(pairs.get(0), password);
        } else {
            return pool.getJedisCluster(pairs, password);
        }
    }


    public Redis getStandaloneResource(Pair<String, Integer> hostAndPort, String password) {

        String host = hostAndPort.getFirst();
        int port = hostAndPort.getSecond();

        if (RedisConnectionProxyConfig.getInstance().isOpen()) {
            try {
                return createJedisBySSH(host, port, password);
            } catch (JSchException e) {
                LogKit.error(e.getMessage(), e);
            }
        } else {
            String feature = host + ":" + port + "@" + password;
            JedisPool jedisPool = jedisPoolMap.get(feature);
            if (jedisPool == null) {
                jedisPool = createJedisPool(host, port, password);
                jedisPoolMap.put(feature, jedisPool);
            }
            return new SimpleRedis(new StandaloneRedisClient(jedisPool.getResource()));
        }
        Jedis jedis = new Jedis(host, port);
        jedis.auth(password);
        return new SimpleRedis(new StandaloneRedisClient(jedis));
    }

    private static JedisPool createJedisPool(String host, int port, String password) {
        RedisConnectionPoolConfig poolConfig = RedisConnectionPoolConfig.getInstance();
        JedisPoolConfig config = new JedisPoolConfig();
        config.setMaxTotal(poolConfig.getMaxTotal());
        //连接耗尽时是否阻塞, false报异常,true阻塞直到超时, 默认true
        config.setBlockWhenExhausted(poolConfig.getBlockWhenExhausted());
        //是否启用pool的jmx管理功能, 默认true
        config.setJmxEnabled(true);
        //MBean ObjectName = new ObjectName("org.apache.commons.pool2:type=GenericObjectPool,name=" + "pool" + i); 默认为"pool", JMX不熟,具体不知道是干啥的...默认就好.
        config.setJmxNamePrefix("pool");
        //是否启用后进先出, 默认true
        config.setLifo(poolConfig.getLifo());
        //最大空闲连接数
        config.setMaxIdle(poolConfig.getMaxIdle());
        //最大连接数，可配置
        //获取连接时的最大等待毫秒数(如果设置为阻塞时BlockWhenExhausted),如果超时就抛异常, 小于零:阻塞不确定的时间,  默认-1
        config.setMaxWaitMillis(poolConfig.getMaxWait());
        //逐出连接的最小空闲时间 默认1800000毫秒(30分钟)
        config.setMinEvictableIdleTimeMillis(1800000);
        //最小空闲连接数, 默认0
        config.setMinIdle(0);
        //每次逐出检查时 逐出的最大数目 如果为负数就是 : 1/abs(n), 默认3
        config.setNumTestsPerEvictionRun(3);
        //对象空闲多久后逐出, 当空闲时间>该值 且 空闲连接>最大空闲数 时直接逐出,不再根据MinEvictableIdleTimeMillis判断  (默认逐出策略)
        config.setSoftMinEvictableIdleTimeMillis(1800000);
        //在获取连接的时候检查有效性, 默认false
        config.setTestOnBorrow(false);
        //在空闲时检查有效性, 默认false
        config.setTestWhileIdle(false);
        //逐出扫描的时间间隔(毫秒) 如果为负数,则不运行逐出线程, 默认-1
        config.setTimeBetweenEvictionRunsMillis(-1);
        if (StringKit.isNotBlank(password)) {
            return new JedisPool(config, host, port, poolConfig.getTimeout(), password);
        } else {
            return new JedisPool(config, host, port, poolConfig.getTimeout());
        }
    }

    private Redis createJedisBySSH(String host, int port, String password) throws JSchException {
        RedisConnectionProxyConfig proxyConfig = RedisConnectionProxyConfig.getInstance();
        JSch jsch = new JSch();
        if (StringKit.isNotBlank(proxyConfig.getPrivateKeyPath())) {
            jsch.addIdentity(proxyConfig.getPrivateKeyPath());
        }
        Session session = jsch.getSession(proxyConfig.getUsername(), proxyConfig.getHost(), proxyConfig.getPort());
        if (StringKit.isNotBlank(proxyConfig.getPassword())) {
            session.setPassword(proxyConfig.getPassword());
        }
        session.setConfig("StrictHostKeyChecking", "no");
        session.connect();

        int newPort = session.setPortForwardingL(0, host, port);

        Jedis jedis = new Jedis(host, newPort);

        if (StringKit.isNotBlank(password)) {
            jedis.auth(password);
        }
        return new ProxyRedis(session, new StandaloneRedisClient(jedis));
    }

    private Redis getJedisCluster(List<Pair<String, Integer>> hostAndPorts, String password) {
        RedisConnectionPoolConfig poolConfig = RedisConnectionPoolConfig.getInstance();
        GenericObjectPoolConfig config = new GenericObjectPoolConfig();

        config.setMaxTotal(poolConfig.getMaxTotal());
        //控制一个pool最多有多少个状态为idle(空闲的)的jedis实例。

        //是否启用后进先出, 默认true
        config.setLifo(poolConfig.getLifo());

        config.setMinIdle(1);
        config.setMaxIdle(poolConfig.getMaxIdle());
        //表示当borrow(引入)一个jedis实例时，最大的等待时间，如果超过等待时间，则直接抛出JedisConnectionException；
        config.setMaxWaitMillis(poolConfig.getMaxWait());
        config.setTimeBetweenEvictionRunsMillis(-1);
        // 在borrow一个jedis实例时，是否提前进行alidate操作；如果为true，则得到的jedis实例均是可用的；
        config.setTestOnBorrow(true);
        // 在还会给pool时，是否提前进行validate操作
        config.setTestOnReturn(true);
        //如果为true，表示有一个idle object evitor线程对idle object进行扫描，如果validate失败，此object会被从pool中drop掉；这一项只有在timeBetweenEvictionRunsMillis大于0时才有意义；
        config.setTestWhileIdle(true);
        //表示一个对象至少停留在idle状态的最短时间，然后才能被idle object evitor扫描并驱逐；这一项只有在timeBetweenEvictionRunsMillis大于0时才有意义；
        config.setMinEvictableIdleTimeMillis(6000);
        //表示idle object evitor两次扫描之间要sleep的毫秒数
        config.setTimeBetweenEvictionRunsMillis(30000);
        Set<HostAndPort> nodes = new LinkedHashSet<HostAndPort>();
        for (Pair<String, Integer> pair : hostAndPorts) {
            nodes.add(new HostAndPort(pair.getFirst(), pair.getSecond()));
        }
        JedisCluster jedisCluster;
        if (StringKit.isNotEmpty(password)) {
            jedisCluster = new JedisCluster(nodes, poolConfig.getTimeout(), 10000, 5, password, config);
        } else {
            jedisCluster = new JedisCluster(nodes, poolConfig.getTimeout(), 10000, 5, config);
        }
        return new SimpleRedis(new ClusterRedisClient(jedisCluster));
    }

    private static List<Pair<String, Integer>> findAllHostAndPassword(String host, String port) {
        String[] hostArray = host.split(",");
        String[] portArray = port.split(",");
        int len = ArrayKit.getLength(hostArray);
        List<Pair<String, Integer>> pairs = new ArrayList<Pair<String, Integer>>();
        for (int i = 0; i < len; i++) {
            pairs.add(new Pair<String, Integer>(hostArray[i], Integer.parseInt(portArray[i])));
        }
        return pairs;
    }
}