redis读写分离及可用性设计

Redis缓存架构设计

对于下面两个架构图，有如下想法：

1）redis主从复制模式，为了解决master读写压力，对master进行写操作，对slave进行读操作。

2）而在分片集群中，如果对部分分片进行写，部分分片进行读，那么会导致写入后无法get指定key的情况。

3）二级缓存有必要吗？二级缓存最主要的问题解决存储介质由磁盘存储转变为内存存储，而redis本身就作为内存数据库，最主要的只能够解决网络问题。

其次多个应用对同一条key-value做修改，是否能保证各个二级缓存数据的一致性呢。

主从读写分离设计

//定义统一连接接口
public interface IRedisConnection {
    public String connection_name = "";
    /**
     * @return the connection_name
     */
    public String getConnection_name() ;
    /**
     * @param connectionName the connection_name to set
     */
    public void setConnection_name(String connectionName) ;
    /**
     * @param key value
     * @param value
     */
    public void write(String key,String value)throws Exception;
    /**
     * @param key
     * @return
     */
    public Object read(String key)throws Exception;
    
//    public boolean isValiate() ;
}

//读连接
public class ReadConnection  extends Jedis implements IRedisConnection{
    public String connection_name = "";
    /**
     * @param host
     */
    public ReadConnection(String host,long port) {
        super(host,(int)port);
        // TODO Auto-generated constructor stub
    }
    /**
     * @return the connection_name
     */
    public String getConnection_name() {
        return connection_name;
    }
    /**
     * @param connectionName the connection_name to set
     */
    public void setConnection_name(String connectionName) {
        connection_name = connectionName;
    }
    
    public Object read(String key){
        return this.get(key);
    }
    public void write(String key,String value) throws Exception{
        throw new Exception("未定义的方法");
    }
}

//写连接
public class WriteConnection  extends Jedis implements IRedisConnection{
    public String connection_name = "";
    /**
     * @param host
     */
    public WriteConnection(String host,long port) {
        super(host,(int)port);
        // TODO Auto-generated constructor stub
    }
    /**
     * @return the connection_name
     */
    public String getConnection_name() {
        return connection_name;
    }
    /**
     * @param connectionName the connection_name to set
     */
    public void setConnection_name(String connectionName) {
        connection_name = connectionName;
    }
    
    public void write(String key,String value){
        this.set(key, value);
    }
    public Object read(String key)throws Exception{
        throw new Exception("未锟斤拷锟斤拷姆锟斤拷锟?");
    }
    
}

//获取一个指定读/写的连接  当然可以，故障检测 ，故障隔离  ，使用 JedisPool保存在内存中
public static IRedisConnection getConnection(String code,
            int connectionType) throws Exception {
        IRedisBaseSV sv = (IRedisBaseSV) ServiceFactory
                .getService(IRedisBaseSV.class);
        CFG_REDIS_SERVERBean server = sv.getServerFromCode(code);
        if (server == null) {
            log.error("根据编码：" + code + "不能找到对应的redis服务器资源");
            throw new Exception("根据编码：" + code + "不能找到对应的redis服务器资源");
        }
        IRedisConnection conn;

        if (connectionType == RedisConstants.WRITE_ONLY_CONNECTION) {
            conn = new WriteConnection(server.getServerIp(), server
                    .getServerPort());
        } else {
            conn = new ReadConnection(server.getServerIp(), server
                    .getServerPort());
        }
        
        conn.setConnection_name(code);
        return conn;
    }

 初始化切片池

/**
   * 初始化切片池
   */
  private static void initialShardedPool() throws Exception {
    long start = System.nanoTime();

    IRedisBaseSV sv = (IRedisBaseSV) ServiceFactory.getService(IRedisBaseSV.class);
    // 初始化JedisPoolConfig
    CFG_REDIS_PARAMETERBean[] para = sv.getRedisConfig("DEFAULT");
    for (int i = 0; i < para.length; i++) {
      if (para[i].getParameterName().equalsIgnoreCase(RedisConstants.REDIS_SERVER_SHARED_MAXACTIVE))
        config.setMaxActive(Integer.parseInt(para[i].getParameterValue()));
      if (para[i].getParameterName().equalsIgnoreCase(RedisConstants.REDIS_SERVER_SHARED_MAXIDLE))
        config.setMaxIdle(Integer.parseInt(para[i].getParameterValue()));
      if (para[i].getParameterName().equalsIgnoreCase(RedisConstants.REDIS_SERVER_SHARED_MAXWAIT))
        config.setMaxWait(Long.parseLong(para[i].getParameterValue()));
      if (para[i].getParameterName().equalsIgnoreCase(
          RedisConstants.REDIS_SERVER_SHARED_TESTONBORROW))
        config.setTestOnBorrow(Boolean.getBoolean(para[i].getParameterValue()));

      if (para[i].getParameterName().equalsIgnoreCase(RedisConstants.REDIS_SERVER_SHARED_NEED_SYN))
        NEED_WRITE_SYNCHRONIZE = Boolean.parseBoolean(para[i].getParameterValue()); // Boolean.getBoolean不对
    }

    CFG_REDIS_SERVERBean[] servers = sv.getRedisServer();
    HashMap map = new HashMap();
    for (int i = 0; i < servers.length; i++) {
      map.put(servers[i].getBelongGroup().toUpperCase(), servers[i].getBelongGroup().toUpperCase());
    }
    serverGroupCodeList = new ArrayList(map.values());
    ShardedJedisPool[] readConnectionPools;
    ShardedJedisPool[] writeConnectionPools;
    ShardedJedisPool[] persistConnectionPools;
    readConnectionPools = new ShardedJedisPool[serverGroupCodeList.size()];
    writeConnectionPools = new ShardedJedisPool[serverGroupCodeList.size()];
    persistConnectionPools = new ShardedJedisPool[serverGroupCodeList.size()];
    for (int i = 0; i < serverGroupCodeList.size(); i++) {
      List<JedisShardInfo> readShards = new ArrayList<JedisShardInfo>();
      List<JedisShardInfo> writeShards = new ArrayList<JedisShardInfo>();
      List<JedisShardInfo> persistShards = new ArrayList<JedisShardInfo>();
      //遍历所有的redis server实例
      for (int j = 0; j < servers.length; j++) {
        if (servers[j].getBelongGroup()
            .equalsIgnoreCase(String.valueOf(serverGroupCodeList.get(i)))
            && servers[j].getUseType().equalsIgnoreCase(
                String.valueOf(RedisConstants.READ_ONLY_CONNECTION))) {
          // 先测试该连接是否可用
          readServersAll.add(servers[j]);
          Boolean connectionTest =
              TestConnection(servers[j].getServerIp(), (int) servers[j].getServerPort(),
                  decryption(servers[j].getRequirepass()), servers[j].getBelongGroup());
          if (connectionTest) {
            JedisShardInfo jds =
                new JedisShardInfo(servers[j].getServerIp(), (int) servers[j].getServerPort());
            if (null != servers[j].getRequirepass()
                && !"".equals(servers[j].getRequirepass().trim()))
              jds.setPassword(decryption(servers[j].getRequirepass()));
            readShards.add(jds);

          }
        } else if (servers[j].getBelongGroup().equalsIgnoreCase(
            String.valueOf(serverGroupCodeList.get(i)))
            && servers[j].getUseType().equalsIgnoreCase(
                String.valueOf(RedisConstants.WRITE_ONLY_CONNECTION))) {
          Boolean connectionTest =
              TestConnection(servers[j].getServerIp(), (int) servers[j].getServerPort(),
                  decryption(servers[j].getRequirepass()), servers[j].getBelongGroup());
          if (connectionTest) {
            JedisShardInfo jds =
                new JedisShardInfo(servers[j].getServerIp(), (int) servers[j].getServerPort());
            if (null != servers[j].getRequirepass()
                && !"".equals(servers[j].getRequirepass().trim()))
              jds.setPassword(decryption(servers[j].getRequirepass()));
            writeShards.add(jds);
          }
        } else if (servers[j].getBelongGroup().equalsIgnoreCase(
            String.valueOf(serverGroupCodeList.get(i)))
            && servers[j].getUseType().equalsIgnoreCase(
                String.valueOf(RedisConstants.PERSIST_ONLYL_CONNECTION))) {
          Boolean connectionTest =
              TestConnection(servers[j].getServerIp(), (int) servers[j].getServerPort(),
                  decryption(servers[j].getRequirepass()), servers[j].getBelongGroup());
          if (connectionTest) {
            JedisShardInfo jds =
                new JedisShardInfo(servers[j].getServerIp(), (int) servers[j].getServerPort());
            if (null != servers[j].getRequirepass()
                && !"".equals(servers[j].getRequirepass().trim()))
              jds.setPassword(decryption(servers[j].getRequirepass()));
            persistShards.add(jds);
          }
        }
      }

      // 构造池    每个组分别对应三个分片池，可读分片池，可写分片池，持久化分片池
      readConnectionPools[i] =
          new ShardedJedisPool(config, readShards, Hashing.MURMUR_HASH,
              Sharded.DEFAULT_KEY_TAG_PATTERN);
      writeConnectionPools[i] =
          new ShardedJedisPool(config, writeShards, Hashing.MURMUR_HASH,
              Sharded.DEFAULT_KEY_TAG_PATTERN);
      persistConnectionPools[i] =
          new ShardedJedisPool(config, persistShards, Hashing.MURMUR_HASH,
              Sharded.DEFAULT_KEY_TAG_PATTERN);
      //按照组名  分别存可读池，可写池，持久化池于对应的map中【此处如果拿到一个可读池，依然可以做写操作】
      readPoolMap.put(serverGroupCodeList.get(i), readConnectionPools[i]);
      writePoolMap.put(serverGroupCodeList.get(i), writeConnectionPools[i]);
      persistPoolMap.put(serverGroupCodeList.get(i), persistConnectionPools[i]);
    }
    long end = System.nanoTime();
    log.debug("初始化连接池用时：" + (end - start));

  }

释放回连接池和销毁连接

  /**
   * 根据组将连接释放，重新放回连接池,解决多个线程频繁取获连接导致连接池连接数不够用的问题
   * 
   * @para gourp 组名
   * @para connectionType 连接类型：1.读2.写
   */
  public static void releaseConnection(ShardedJedis jedis, String group, int connectionType)
      throws Exception {
    try {
      ShardedJedisPool rp = null;
      if (connectionType == RedisConstants.READ_ONLY_CONNECTION) {
        rp = (ShardedJedisPool) readPoolMap.get(group.toUpperCase());
      } else if (connectionType == RedisConstants.WRITE_ONLY_CONNECTION) {
        rp = (ShardedJedisPool) writePoolMap.get(group.toUpperCase());
      } else if (connectionType == RedisConstants.PERSIST_ONLYL_CONNECTION) {
        rp = (ShardedJedisPool) persistPoolMap.get(group.toUpperCase());
      }
      rp.returnResource(jedis);
    } catch (Exception ex) {
      log.error("释放连接debug");
    }
  }

  /**
   * 销毁连接 连接超时以后不能只是释放连接，需要销毁，否则下次使用会取到上次的结果，出现类型转换出错的问题。
   * 
   * @param jedis
   * @param group
   * @param connectionType
   * @throws Exception
   */
  public static void destoryConnection(ShardedJedis jedis, String group, int connectionType)
      throws Exception {
    try {
      ShardedJedisPool rp = null;
      if (connectionType == RedisConstants.READ_ONLY_CONNECTION) {
        rp = (ShardedJedisPool) readPoolMap.get(group.toUpperCase());
      } else if (connectionType == RedisConstants.WRITE_ONLY_CONNECTION) {
        rp = (ShardedJedisPool) writePoolMap.get(group.toUpperCase());
      } else if (connectionType == RedisConstants.PERSIST_ONLYL_CONNECTION) {
        rp = (ShardedJedisPool) persistPoolMap.get(group.toUpperCase());
      }
      rp.returnBrokenResource(jedis);
    } catch (Exception ex) {
      log.error("销毁连接debug");
    }
  }

 测试连接，可用于故障隔离

  /**
   * 初始化时调用，用于初始化set初始值，并测试连通性
   * @param host
   * @param port
   * @param passwd
   * @param group
   * @return
   */
  private static Boolean TestConnection(String host, int port, String passwd, String group) {
    Boolean rtn = Boolean.FALSE;
    JedisShardInfo jsd = new JedisShardInfo(host, port);
    if (null != passwd && !"".equals(passwd.trim()))
      jsd.setPassword(passwd);
    Jedis jd = null;
    try {
      jd = new Jedis(jsd);
      jd.set(group + "AILK_REDIS_CONNECT_TEST", "TRUE");
      rtn = Boolean.TRUE;
    } catch (Exception ex) {
      if (log.isDebugEnabled()) {
        log.error("【调试】," + host + ":" + port + "拒绝连接！" + ex.getMessage());
      }
    } finally {
      if (null != jd)
        jd.disconnect();
    }
    return rtn;
  }

  /**
   * 故障隔离调用，通过是否能够get值来判断
   * @param host
   * @param port
   * @param passwd
   * @param group
   * @return
   */
  private static Boolean isConnectioned(String host, int port, String passwd, String group) {
    Boolean rtn = Boolean.FALSE;
    JedisShardInfo jsd = new JedisShardInfo(host, port);
    if (null != passwd && !"".equals(passwd.trim()))
      jsd.setPassword(passwd);
    Jedis jd = null;
    try {
      jd = new Jedis(jsd);
      String ailkRedisConnectTest = jd.get(group + "AILK_REDIS_CONNECT_TEST");
      if (ailkRedisConnectTest != null && ailkRedisConnectTest.equalsIgnoreCase("true"))
        rtn = Boolean.TRUE;
    } catch (Exception ex) {
      if (log.isDebugEnabled()) {
        log.error("【调试】," + host + ":" + port + "拒绝连接！" + ex.getMessage());
      }
    } finally {
      if (null != jd)
        jd.disconnect();
    }
    return rtn;
  }