Java 类redis.clients.jedis.ShardedJedisPool 实例源码

@Bean(name="monitorJedisPool")
public ShardedJedisPool getJedisPool() {
    try {
        List<JedisShardInfo> shardList = new ArrayList<>();
        String hostStr = env.getProperty("spring.redis.shard.monitor.hostList");
        int timeOut = Integer.parseInt(env.getProperty("spring.redis.shard.monitor.timeOut"));
        if(!StringUtils.isEmpty(hostStr)){
            String[] hosts = hostStr.split(",");
            for(int i = 0; i < hosts.length; i++){
                String[] temp = hosts[i].split(":");
                JedisShardInfo info = new JedisShardInfo(temp[0], Integer.valueOf(temp[1]), timeOut);
                shardList.add(info);
            }
        }

        if(shardList.size() == 0){
            //无法加载redis
            throw new IOException();
        }
        return new ShardedJedisPool(getJedisPoolConfig(), shardList);
    } catch (Exception e) {
        throw new RuntimeException("无法加载资源文件!");
    }
}

@Override
public void afterPropertiesSet() throws Exception {
    final JedisPoolConfig config = JedisPoolConfigFactory.createJedisPoolConfig();

    final Set<JedisShardInfo> shardInfos = new HashSet<JedisShardInfo>();
    final HostInfo[] hostInfos = HostInfoFactory.split(hosts);
    for (final HostInfo hostInfo : hostInfos) {
        shardInfos.add(hostInfo.createJedisShardInfo());
    }
    if (redisService == null) {
        final ShardedJedisPool jedisPool = new ShardedJedisPool(config, new ArrayList<JedisShardInfo>(shardInfos));
        redisService = new RedisServiceImpl(jedisPool);
    }

    final RedisServiceProxy redisServiceProxy = new RedisServiceProxy();
    this.redisService = redisServiceProxy.bind(redisService);
}

/** 
 * 初始化切片池 
 */ 
private void initialShardedPool() 
{ 
    // 池基本配置 
    JedisPoolConfig config = new JedisPoolConfig(); 
    config.setMaxActive(20); 
    config.setMaxIdle(5); 
    config.setMaxWait(1000l); 
    config.setTestOnBorrow(false); 
    // slave链接 
    List<JedisShardInfo> shards = new ArrayList<JedisShardInfo>(); 
    shards.add(new JedisShardInfo("127.0.0.1", 6379, "master")); 

    // 构造池 
    shardedJedisPool = new ShardedJedisPool(config, shards); 
}

/**
 * 
 * @return
 */
public static JedisManager createJedisManager() {

    Config conf = Config.newInstance();

    // 池基本配置
    JedisPoolConfig config = new JedisPoolConfig();
    config.setMaxIdle(conf.getInt("redis.maxIdle"));
    config.setMinIdle(conf.getInt("redis.minIdle"));
    config.setMaxWaitMillis(conf.getLong("redis.maxWaitMillis"));
    config.setTestOnBorrow(conf.getBoolean("redis.testOnBorrow"));
    List<JedisShardInfo> shards = new ArrayList<JedisShardInfo>();
    // slave链接
    shards.add(new JedisShardInfo(conf.getString("redis.slave.host"), conf
            .getInt("redis.slave.port"), conf.getInt("redis.timeout")));
    // master链接
    shards.add(new JedisShardInfo(conf.getString("redis.master.host"), conf
            .getInt("redis.master.port"), conf.getInt("redis.timeout")));

    // 构造池
    ShardedJedisPool shardedJedisPool = new ShardedJedisPool(config, shards);

    return new JedisManager(shardedJedisPool);
}

/**
 * 根据配置创建
 * 
 * @return
 */
private static JedisManager createJedisManager() {

    // 池基本配置
    JedisPoolConfig config = new JedisPoolConfig();
    config.setMaxIdle(Conf.getInt("redis.maxIdle"));
    config.setMinIdle(Conf.getInt("redis.minIdle"));
    config.setMaxWaitMillis(Conf.getLong("redis.maxWaitMillis"));
    config.setTestOnBorrow(Conf.getBoolean("redis.testOnBorrow"));
    List<JedisShardInfo> shards = new ArrayList<JedisShardInfo>();
    // 链接
    shards.add(new JedisShardInfo(Conf.getString("redis.host"), Conf.getInt("redis.port"), Conf
            .getInt("redis.timeout")));

    // 构造池
    ShardedJedisPool shardedJedisPool = new ShardedJedisPool(config, shards);

    return new JedisManager(shardedJedisPool);
}

/**
 * 根据配置创建
 * 
 * @return
 */
private static JedisManager createJedisManager() {

    // 池基本配置
    JedisPoolConfig config = new JedisPoolConfig();
    config.setMaxIdle(Conf.getInt("redis.maxIdle"));
    config.setMinIdle(Conf.getInt("redis.minIdle"));
    config.setMaxWaitMillis(Conf.getLong("redis.maxWaitMillis"));
    config.setTestOnBorrow(Conf.getBoolean("redis.testOnBorrow"));
    List<JedisShardInfo> shards = new ArrayList<JedisShardInfo>();
    // 链接
    shards.add(new JedisShardInfo(Conf.getString("redis.host"), Conf.getInt("redis.port"), Conf
            .getInt("redis.timeout")));

    // 构造池
    ShardedJedisPool shardedJedisPool = new ShardedJedisPool(config, shards);

    return new JedisManager(shardedJedisPool);
}

public ShardedJedisPool appendJedis(final RedisConfig conf) {
    Assert.notNull(conf);
    Assert.hasLength(conf.getRedisType());

    if(conf.getCluster() == null || !conf.getCluster()) {
        if (!jedisPool.containsKey(conf.getRedisType())) {
            redisConfigs.put(conf.getRedisType(), conf);
            final ShardedJedisPool pool;
            jedisPool.put(conf.getRedisType(), pool = createJedisPool(conf));
            bindGlobal(conf);
            return pool;
        }

        return jedisPool.get(conf.getRedisType());
    }

    throw new RedisClientException("Can't append ShardedJedis, this is a redis cluster config");
}

public ShardedJedis getJedis(final String poolName) {
    Assert.hasText(poolName);
    ShardedJedis shardedJedis = null;
    try {
        final ShardedJedisPool pool = jedisPool.get(poolName);
        if (pool != null) {
            shardedJedis = pool.getResource();
        }

        Assert.notNull(shardedJedis, "Not found ShardedJedis.");
        return shardedJedis;
    } catch (final Throwable e) {
        close(shardedJedis);
        throw new RedisClientException(e.getMessage(), e);
    }
}

public void create() {
    try {
        LOG.debug("Starting RedisShardedProvider ... configuration:" + redisConfiguration);

        JedisPoolConfig poolConfig = new JedisPoolConfig();
        poolConfig.setMaxTotal(1000);
        poolConfig.setMinIdle(2);

        pool = new ShardedJedisPool(poolConfig, shards(redisConfiguration.getServers()));

        testConnection();
        LOG.debug("RedisShardedProvider started.");
    } catch (Exception e) {
        LOG.error("Failed to start RedisShardedProvider.");
        throw new IllegalStateException("Error starting RedisShardedProvider", e);
    }
}

private void init() {
    redis1.slaveofnoone();
    redis2.slaveofnoone();
    redis3.slaveofnoone();
    List<Redis> shards = new ArrayList<Redis>(3);
    shards.add(redis1);
    shards.add(redis2);
    shards.add(redis3);
    shardedRedis = RedisFactory.newShardedRedis(shards);

    List<JedisShardInfo> shardinfos = new ArrayList<JedisShardInfo>();
    shardinfos.add(new JedisShardInfo(HOST, PORT1));
    shardinfos.add(new JedisShardInfo(HOST, PORT2));
    shardinfos.add(new JedisShardInfo(HOST, PORT3));
    shardedPool = new ShardedJedisPool(new JedisPoolConfig(), shardinfos);
}

/**
 * @param args
 */
public static void main(String[] args) {
    List<JedisShardInfo> list = new LinkedList<JedisShardInfo>();
    JedisShardInfo jedisShardInfo1 = new JedisShardInfo(ip1, port);
    jedisShardInfo1.setPassword(JedisConstant.password);
    list.add(jedisShardInfo1);
    JedisShardInfo jedisShardInfo2 = new JedisShardInfo(ip2, port);
    jedisShardInfo2.setPassword(JedisConstant.password);
    list.add(jedisShardInfo2);
    ShardedJedisPool pool = new ShardedJedisPool(config, list);
    for (int i = 0; i < 2000; i++) {
        ShardedJedis jedis = pool.getResource();
        String key = "howsun_" + i;
        //jedis.set(key, UUID.randomUUID().toString());
        System.out.println(key + "\t" + jedis.get(key) + "\t" + jedis.toString());
        pool.returnResource(jedis);
    }
}

/**
 * Create a new {@link ShardedJedisPool}.
 * 
 * @param poolConfig
 *            format {@code host1:port1,host2:port2,...}, default Redis port is used if not
 *            specified
 * @param password
 * @param timeoutMs
 * @return
 */
public static ShardedJedisPool newShardedJedisPool(JedisPoolConfig poolConfig,
        String hostsAndPorts, String password, int timeoutMs) {
    List<JedisShardInfo> shards = new ArrayList<>();
    String[] hapList = hostsAndPorts.split("[,;\\s]+");
    for (String hostAndPort : hapList) {
        String[] tokens = hostAndPort.split(":");
        String host = tokens.length > 0 ? tokens[0] : Protocol.DEFAULT_HOST;
        int port = tokens.length > 1 ? Integer.parseInt(tokens[1]) : Protocol.DEFAULT_PORT;
        JedisShardInfo shardInfo = new JedisShardInfo(host, port, timeoutMs);
        shardInfo.setPassword(password);
        shards.add(shardInfo);
    }
    ShardedJedisPool jedisPool = new ShardedJedisPool(poolConfig, shards);
    return jedisPool;
}

public static void setSharedJedisPool() {
        JedisPoolConfig config = new JedisPoolConfig();//Jedis池配置
        // 共享jedis池，用于集群
        List<JedisShardInfo> jedisShardInfos = new ArrayList<>();
        JedisShardInfo node01 = new JedisShardInfo("120.24.238.195", 6379);
//        JedisShardInfo node02 = new JedisShardInfo("120.25.162.32", 6379);
//        JedisShardInfo node03 = new JedisShardInfo("112.74.114.226", 6379);
        node01.setPassword("110110");
        jedisShardInfos.add(node01);
//        jedisShardInfos.add(node02);
//        jedisShardInfos.add(node03);
        pool = new ShardedJedisPool(config, jedisShardInfos);
        isSetup = true;
    }

@Test
public void checkConnections() {
  ShardedJedisPool pool = new ShardedJedisPool(new GenericObjectPoolConfig(), shards);
  ShardedJedis jedis = pool.getResource();
  jedis.set("foo", "bar");
  assertEquals("bar", jedis.get("foo"));
  pool.returnResource(jedis);
  pool.destroy();
}

@Test
public void checkCloseableConnections() throws Exception {
  ShardedJedisPool pool = new ShardedJedisPool(new GenericObjectPoolConfig(), shards);
  ShardedJedis jedis = pool.getResource();
  jedis.set("foo", "bar");
  assertEquals("bar", jedis.get("foo"));
  pool.returnResource(jedis);
  pool.close();
  assertTrue(pool.isClosed());
}

@Test
public void checkConnectionWithDefaultPort() {
  ShardedJedisPool pool = new ShardedJedisPool(new GenericObjectPoolConfig(), shards);
  ShardedJedis jedis = pool.getResource();
  jedis.set("foo", "bar");
  assertEquals("bar", jedis.get("foo"));
  pool.returnResource(jedis);
  pool.destroy();
}

@Test
public void checkJedisIsReusedWhenReturned() {
  ShardedJedisPool pool = new ShardedJedisPool(new GenericObjectPoolConfig(), shards);
  ShardedJedis jedis = pool.getResource();
  jedis.set("foo", "0");
  pool.returnResource(jedis);

  jedis = pool.getResource();
  jedis.incr("foo");
  pool.returnResource(jedis);
  pool.destroy();
}

@Test
public void checkPoolRepairedWhenJedisIsBroken() {
  ShardedJedisPool pool = new ShardedJedisPool(new GenericObjectPoolConfig(), shards);
  ShardedJedis jedis = pool.getResource();
  jedis.disconnect();
  pool.returnBrokenResource(jedis);

  jedis = pool.getResource();
  jedis.incr("foo");
  pool.returnResource(jedis);
  pool.destroy();
}

@Test(expected = JedisException.class)
public void checkPoolOverflow() {
  GenericObjectPoolConfig config = new GenericObjectPoolConfig();
  config.setMaxTotal(1);
  config.setBlockWhenExhausted(false);

  ShardedJedisPool pool = new ShardedJedisPool(config, shards);

  ShardedJedis jedis = pool.getResource();
  jedis.set("foo", "0");

  ShardedJedis newJedis = pool.getResource();
  newJedis.incr("foo");
}

@Test
public void shouldNotShareInstances() {
  GenericObjectPoolConfig config = new GenericObjectPoolConfig();
  config.setMaxTotal(2);

  ShardedJedisPool pool = new ShardedJedisPool(config, shards);

  ShardedJedis j1 = pool.getResource();
  ShardedJedis j2 = pool.getResource();

  assertNotSame(j1.getShard("foo"), j2.getShard("foo"));
}

@Test
public void checkFailedJedisServer() {
  ShardedJedisPool pool = new ShardedJedisPool(new GenericObjectPoolConfig(), shards);
  ShardedJedis jedis = pool.getResource();
  jedis.incr("foo");
  pool.returnResource(jedis);
  pool.destroy();
}

public static void main(String[] args) {
    JedisPoolConfig jedisPoolConfig = new JedisPoolConfig();
    jedisPoolConfig.setMaxTotal(50000);
    jedisPoolConfig.setMaxIdle(20000);
    jedisPoolConfig.setMaxWaitMillis(3000);
    jedisPoolConfig.setTestOnBorrow(true);
    jedisPoolConfig.setTestOnReturn(true);

    List<JedisShardInfo> shards = new ArrayList<>();
    shards.add(new JedisShardInfo("127.0.0.1", 6379, 300));
    ShardedJedisPool shardedJedisPool = new ShardedJedisPool(jedisPoolConfig, shards);
    RedisService redisService = new RedisService(shardedJedisPool);
    new Thread(new HallTcpService(redisService)).start();
    new Thread(new HallNoticeService(redisService)).start();
}

private static ShardedJedis getJedis() {
    if (shardedJedisPool == null) {
        synchronized (EXPIRE) {
            if (shardedJedisPool == null) {
                WebApplicationContext wac = ContextLoader.getCurrentWebApplicationContext();
                shardedJedisPool = wac.getBean(ShardedJedisPool.class);
            }
        }
    }
    return shardedJedisPool.getResource();
}

public void init() {

        try {
            String[] hosts = servers.trim().split("\\|");

            List<JedisShardInfo> shards = new ArrayList<JedisShardInfo>();

            for (String host : hosts) {
                String[] ss = host.split(":");
                //升级 redis  构造变化
                JedisShardInfo shard = new JedisShardInfo(ss[0], Integer.parseInt(ss[1]), DEFAULT_TIMEOUT, DEFAULT_TIMEOUT, 1);
                shards.add(shard);
            }

            JedisPoolConfig config = new JedisPoolConfig();
            config.setMaxIdle(MAX_ACTIVE);
            config.setMinIdle(MAX_IDLE);
            config.setMaxWaitMillis(MAX_WAIT);
            config.setMaxTotal(MAX_TOTAL);

            pool = new ShardedJedisPool(config, shards, Hashing.MURMUR_HASH);
        } catch (NumberFormatException e) {
            System.out.println("redis客户端初始化连接异常!!!!!!!!!  链接参数:" + servers
                    + " " + DEFAULT_TIMEOUT + " " + app);
            logger.error("redis:{},exception:{}.", servers + " "
                    + DEFAULT_TIMEOUT + " " + app, e.getMessage());
        }
    }

@Override
public Pool<ShardedJedis> initPool() throws Exception {
    if (Check.isNullOrEmpty(getServers())) {
        throw new IllegalArgumentException("未指定redis服务器地址");
    }
    String[] hosts = getServers().trim().split("\\|");
    List<JedisShardInfo> shards = new ArrayList<JedisShardInfo>();
    for (String host : hosts) {
        String[] ss = host.split(":");
        //升级 redis  构造变化
        JedisShardInfo shard = new JedisShardInfo(ss[0], Integer.parseInt(ss[1]), connectTimeout, socketTimeout, 1);
        shards.add(shard);
    }
    return new ShardedJedisPool(getPoolConfig(), shards, Hashing.MURMUR_HASH);
}

public JedisShardProvider(String groupName, JedisPoolConfig jedisPoolConfig, String[] servers,
                          int timeout) {
    super();
    this.groupName = groupName;
    List<JedisShardInfo> shards = buildShardInfos(servers, timeout);
    jedisPool = new ShardedJedisPool(jedisPoolConfig, shards);
}

@BeforeClass
public static void initialize() {
    List<JedisShardInfo> shards = new ArrayList<JedisShardInfo>();
    JedisShardInfo shard1 = new JedisShardInfo(TestConfig.HOST_NAME, 16379);
    shards.add(shard1);

    JedisShardInfo shard2 = new JedisShardInfo(TestConfig.HOST_NAME, 16479);
    shards.add(shard2);

    JedisShardInfo shard3 = new JedisShardInfo(TestConfig.HOST_NAME, 16579);
    shards.add(shard3);

    jedisPool = new ShardedJedisPool(new GenericObjectPoolConfig(), shards);
}

/**
 * redis分片缓存
 * @param shardedJedisPool 分片池
 * @param reConnectPercent 百分比
 * @param reConnectTimeout 重连毫秒值
 */
public ShardedJedisCache(ShardedJedisPool shardedJedisPool, int reConnectPercent, int reConnectTimeout) {
    super();
    this.shardedJedis = shardedJedisPool.getResource();
    this.reConnectPercent = reConnectPercent;
    this.reConnectTimeout = reConnectTimeout;
}

/**
 * Inits the.
 *
 * @param conf
 *          the conf
 * @return the i cache system
 */
public static ICacheSystem create(String server) {

  RedisCache r = new RedisCache();
  r.url = server.substring(Cache.REDIS.length());

  String[] ss = r.url.split(":");
  String host = ss[0];
  int port = 6379;
  if (ss.length > 1) {
    port = X.toInt(ss[1], 0);
  }

  JedisPoolConfig config = new JedisPoolConfig();
  config.setMaxTotal(20);
  config.setMaxIdle(5);
  config.setMaxWaitMillis(1000l);
  config.setTestOnBorrow(false);

  List<JedisShardInfo> shards = new ArrayList<JedisShardInfo>();
  shards.add(new JedisShardInfo(host, port, "master"));

  r.shardedJedisPool = new ShardedJedisPool(config, shards);
  r.jedis = r.shardedJedisPool.getResource();

  return r;
}

@Test
public void checkConnections() {
  ShardedJedisPool pool = new ShardedJedisPool(new GenericObjectPoolConfig(), shards);
  ShardedJedis jedis = pool.getResource();
  jedis.set("foo", "bar");
  assertEquals("bar", jedis.get("foo"));
  jedis.close();
  pool.destroy();
}

@Test
public void checkCloseableConnections() throws Exception {
  ShardedJedisPool pool = new ShardedJedisPool(new GenericObjectPoolConfig(), shards);
  ShardedJedis jedis = pool.getResource();
  jedis.set("foo", "bar");
  assertEquals("bar", jedis.get("foo"));
  jedis.close();
  pool.close();
  assertTrue(pool.isClosed());
}

@Test
public void checkConnectionWithDefaultPort() {
  ShardedJedisPool pool = new ShardedJedisPool(new GenericObjectPoolConfig(), shards);
  ShardedJedis jedis = pool.getResource();
  jedis.set("foo", "bar");
  assertEquals("bar", jedis.get("foo"));
  jedis.close();
  pool.destroy();
}

@Test
public void checkJedisIsReusedWhenReturned() {
  ShardedJedisPool pool = new ShardedJedisPool(new GenericObjectPoolConfig(), shards);
  ShardedJedis jedis = pool.getResource();
  jedis.set("foo", "0");
  jedis.close();

  jedis = pool.getResource();
  jedis.incr("foo");
  jedis.close();
  pool.destroy();
}

@Test
public void checkPoolRepairedWhenJedisIsBroken() {
  ShardedJedisPool pool = new ShardedJedisPool(new GenericObjectPoolConfig(), shards);
  ShardedJedis jedis = pool.getResource();
  jedis.disconnect();
  jedis.close();

  jedis = pool.getResource();
  jedis.incr("foo");
  jedis.close();
  pool.destroy();
}