对于下面两个架构图，有如下想法：
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; }
本站文章如无特殊说明，均为本站原创，如若转载，请注明出处：redis读写分离及可用性设计 - Python技术站
微信扫一扫 
支付宝扫一扫 