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>() { @Override public boolean match(Class clazz) { return RedisConnectionPoolConfig.class == clazz; } }, new Runner() { @Override public void run() { pool.clearAll(); } }); } private Map jedisPoolMap = new ConcurrentHashMap(); private void clearAll() { jedisPoolMap.clear(); } public static Redis getFinal(String host, String port, String password) { List> pairs = findAllHostAndPassword(host, port); if (pairs.size() == 1) { return pool.getStandaloneResource(pairs.get(0), password); } else { return pool.getJedisCluster(pairs, password); } } public Redis getStandaloneResource(Pair 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> 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 nodes = new LinkedHashSet(); for (Pair 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> findAllHostAndPassword(String host, String port) { String[] hostArray = host.split(","); String[] portArray = port.split(","); int len = ArrayKit.getLength(hostArray); List> pairs = new ArrayList>(); for (int i = 0; i < len; i++) { pairs.add(new Pair(hostArray[i], Integer.parseInt(portArray[i]))); } return pairs; } }