Redis - Springboot中集成多个Redis客户端统一管理

Posted 2022-01-08 Zyred

Springboot中集成多个Redis客户端统一管理

前言：　我们都知道 springboot 中要么是集群管理，要么是对单个 Redis 管理，但是维度没有对多个 Redis 管理。在工作中我们都发现 Redis 如果是用集群存储，那每个 Redis 服务器上保存的数据都是相同，其实这并不是我想要的，而我想要的是每个 Redis 上存储的内容是不相同的。例如 Redis1 上存储 1、2、3 的 key，Redis2 上存储 4、5、6 的数据，客户端只通过一套统一的代码获取到对应的值，那么这种我们该怎么去处理这样的问题。

代码地址： https://gitee.com/Zyred/multi_conn

1. 思路分析

结合前三篇对 HashMap 原理的文章，知道了 HashMap 的存储原理是通过一个数组，然后根据 K 的 hash 值计算出 V 存储在数组哪个下标下，有了这种思路我们就可以开始创作。那么我们在来回顾一下 HashMap 原理图:

通过图中不难发现，我的思路无非就是没有链表和红黑树而已

2. 准备工作

• 一个配置文件：保存对 Redis 服务器的配置，而且要方便解析，最终采用 properties 类型的文件
• 解析配置文件工具类 ：对 properties 的解析，解析成自己能理解的 JavaBean 对象，方便后续 getter 属性
• 保存连接、获取连接：spring 启动之初将 Redis 连接的全部创建并且将所有的操作 API 对象保存到内存中

3. 思考

问题1：如何解析 Redis 配置文件？

• 采用 yml文件：
  结合 ClassPathResource 对象读取，解析成为 Properties 对象，配置文件格式如下：

multi:
  conn:
    client1: 
      database: 1
      host: localhost
      port: 6379
    client2:
      database: 2
      host: localhost
      port: 6379

通过对以上配置文件的解析，解析出来是存在一定问题的，Properties 文件主要是 K V 的组合，最终无法读取正确的格式

• 采用 properties 文件：
  结合 ClassPathResource 对象读取，解析成为 Properties 对象，配置文件格式如下：

### 客户端1配置
redis.multi.client1.database=1
redis.multi.client1.host=localhost
redis.multi.client1.port=6379
redis.multi.client1.password=
redis.multi.client1.timeout=0
### 客户端2配置
redis.multi.client2.database=2
redis.multi.client2.host=localhost
redis.multi.client2.port=6379
redis.multi.client2.password=
redis.multi.client2.timeout=0

通过这样的方式读取是完全没有问题的，此方式的可行性很高

问题2：如何创建多个 Redis 连接并保存到内存中 ？
到达这一步的前提是必须读取到配置文件，通过配置文件创建出 Redis 连接，在 spring boot 中引入了 spring-boot-starter-data-redis 这一个依赖，里面包含了 RedisTemplate 与 StringRedisTemplate 等的 Redis 操作基类，所以可以利用这一点将操作基类保存到内存中。
那么上面提到了 RedisTemplate 与 StringRedisTemplate 具体应该怎么选择，其实两者都是大同小异的，由于 Redis 中存储 String 类型的数据更方便我们开发人员通过可视化同居直观判断，所以我选择了 StringRedisTemplate，并且这个类也不需要手动设置 key、value、hash ... 等数据类型的序列化器。所以我选择了 StringRedisTemplate，而创建连接则是通过如下代码：

final StringRedisTemplate t = new StringRedisTemplate();

RedisStandaloneConfiguration sc =
        new RedisStandaloneConfiguration(p.getHost(), p.getPort());
sc.setDatabase(p.getDatabase());
sc.setPassword(p.getPassword());

JedisConnectionFactory factory = new JedisConnectionFactory(sc);

t.setConnectionFactory(factory);
// 必须要调用本方法进行初始化每一个连接
t.afterPropertiesSet();

问题3：如何设计出 key 精准定位 val 所在的 Redis 内
谈到这里不得不说一下 HashMap 的原理了，应为我的设计中这里借鉴了 HashMap 的设计原理。HashMap 中是通过计算 key 的 hashCode 值，然后使用 hash 值与本身高 16 位做异或运算降低 hash 冲突，最终返回的 hash 值与数组的 长度 - 1 做与运算得到 key 所在的下标位置。而在我们自己的设计中，恰巧也是可以这么做的，那么代码如下：

int slot(String key, int size) 
    inth;
    return ((key == null) ? 0 : (h = key.hashCode()) ^ (h >>> 16)) & (size - 1);

4. 部分代码

• 加载配置文件，配置文件名称固定，配置文件名称 multi-redis.properties

@Configuration
public class MultiRedisConnProperties 

    private final String profile = "multi-redis.properties", prefix = "redis.multi.client", command = ".";

    private List<RedisConnProperties> redisConnProperties;

    public MultiRedisConnProperties() 
        try 
        	// 初始化
            this.redisConnProperties = new ArrayList<>();
            // 把配置文件转换成 ClassPathResource 对象
            this.loadProfile(new ClassPathResource(profile));
         catch (IOException e) 
            e.printStackTrace();
        
    


    private void loadProfile(Resource resource) throws IOException 
    
        if (!resource.exists()) 
            throw new FileNotFoundException("Profile " + profile + " not found.");
        

        InputStream input = resource.getInputStream();
        // 将ClassPathResource转换为 properties 对象
        Properties p = new Properties();
        p.load(input);

        Set<String> ns = p.stringPropertyNames();
        // redis.multi.client 的长度 + 1
        int len = prefix.length() + 1;

		// 这里能成功获取到 client 的个数
        Set<String> pres = ns.stream().map(a -> a.substring(0, len))
                .collect(Collectors.toSet());

		int size, last = 0;
        if ((size = pres.size()) == 0 || size != (last = calc(size))) 
            throw new MultiException("The number of clients must be 2 to the NTH power, the current number is only " + size +", you can set the number to" + last);
		
		// 循环创建 redis 配置对象
        for (String pre : pres) 
            RedisConnProperties pro = new RedisConnProperties();
            pro.setClientName(pre.substring(pre.lastIndexOf(command) + 1));

            for (String next : ns) 

                String nextPre = next.substring(0, len);
                if (!pre.equals(nextPre)) continue;
				
				// 截取属性名称
                int index = next.lastIndexOf(command) + 1;
                String suffix = next.substring(index);
				
				// 为属性名称通过反射写入值
                this.write(pro, suffix, p.getProperty(next));
            
            // 将配置装入容器中
            this.redisConnProperties.add(pro);
        

    

    private void write (Object target, String field, String val) 
        try 
            Class<?> clazz = target.getClass();
            Field f = clazz.getDeclaredField(field);

            f.setAccessible(true);
            Class<?> t = f.getType();

            Object value = null;
            if (t == Integer.class || t == int.class) 
                value = Integer.parseInt(val);
             else if (t == long.class || t == Long.class) 
                value = Long.parseLong(val);
            
            f.set(target, value == null ? val : value);
         catch (Exception ex) 
            ex.printStackTrace();
        
    

    public final int getSize () 
        return this.redisConnProperties.size();
    

    public final List<RedisConnProperties> getMulti () 
        return this.redisConnProperties;
    

	// 计算客户端的数量是不是满足 2 的 n 次方个数
	// 此处不考虑超过 Integer.MAX，没有谁会无聊到在配置文件中复制粘贴这么多个
    static final int calc (int size)  
        int pre = 1, next = pre << 1;
        for (;;) 

            if (size == pre) 
                return size;
            

            if (size > pre && size < next) 
                return next;
            

            pre = next;
            next = next << 1;
        
    

• 加载配置文件、实现插入和获取值的逻辑

public interface MultiRedisOperation 

    /**
     * 插入一个不带超时时间的值到 redis 中
     * @param k     K
     * @param v     V
     */
    void put(String k, Object v);

    /**
     * 插入一个带超时时间的值到 redis 中
     *
     * @param k         K
     * @param v         V
     * @param timeout   超时时间
     * @param unit      时间单位
     */
    void put(String k, Object v, int timeout, TimeUnit unit);

    /**
     * 获取一个值
     *
     * @param k     K
     * @param clazz 值类型的class对象
     * @param <V>   类型泛型
     * @return      泛型对应的实体
     */
    <V> V get (String k, Class<V> clazz);

    /**
     * 获取一个 string 类型的值
     *
     * @param k     K
     * @return      string 类型的值
     */
    String get (String k);


    static int slot(String key, int size) 
        int h;
        return ((key == null) ? 0 : (h = key.hashCode()) ^ (h >>> 16)) & (size - 1);
    

@Configuration
public class MultiRedisConnectionFactory implements MultiRedisOperation 

    private final MultiRedisConnProperties ps;
    private final Map<Integer, StringRedisTemplate> conns;
    private final int size;

    public MultiRedisConnectionFactory(MultiRedisConnProperties ps) 
        this.ps = ps;
        this.conns = new ConcurrentHashMap<>();
        this.initConn();
        this.size = ps.getSize();
    

    private void initConn() 
        List<RedisConnProperties> multi = this.ps.getMulti();
        for (int i = 0; i < multi.size(); i++) 
            final RedisConnProperties p = multi.get(i);

            final StringRedisTemplate t = new StringRedisTemplate();

            RedisStandaloneConfiguration sc =
                    new RedisStandaloneConfiguration(p.getHost(), p.getPort());
            sc.setDatabase(p.getDatabase());
            sc.setPassword(p.getPassword());

            JedisConnectionFactory factory = new JedisConnectionFactory(sc);

            t.setConnectionFactory(factory);
            // 必须要调用本方法进行初始化每一个连接
            t.afterPropertiesSet();

            conns.put(i, t);
        
    

    @Override
    public void put(String k, Object v) 
        this.putVal(MultiRedisOperation.slot(k, size), k, v);
    

    @Override
    public void put(String k, Object v, int timeout, TimeUnit unit) 
        this.putVal(MultiRedisOperation.slot(k, size), k, v, timeout, unit);
    

    void putVal(int hash, String k, Object v) 
        int slot = hash & size - 1;
        StringRedisTemplate temp = this.conns.get(slot);
        temp.opsForValue().set(k, JSONObject.toJSONString(v));
    

    void putVal(int hash, String k, Object v, int timeout, TimeUnit unit) 
        int slot = hash & size - 1;
        StringRedisTemplate temp = this.conns.get(slot);
        temp.opsForValue().set(k, JSONObject.toJSONString(v), timeout, unit);
    

    @Override
    public <V> V get (String k, Class<V> clazz) 
        String val = this.conns.get(MultiRedisOperation.slot(k, size)).opsForValue().get(k);
        return JSONObject.parseObject(val, clazz);
    

    @Override
    public String get (String k) 
        return this.conns.get(MultiRedisOperation.slot(k, size)).opsForValue().get(k);
    

5. 图解 MultiRedis

6. 总结

该功能还有待完善，本文章中只是完成了我的一个突发奇想，还未真正运用在商业或者项目中，如果有雷同的需求或者有更好的思路的可以留言与我一起讨论，一起学习。