Springboot中集成多个Redis客户端统一管理
前言： 我们都知道 springboot 中要么是集群管理，要么是对单个 Redis 管理，但是维度没有对多个 Redis 管理。在工作中我们都发现 Redis 如果是用集群存储，那每个 Redis 服务器上保存的数据都是相同，其实这并不是我想要的，而我想要的是每个 Redis 上存储的内容是不相同的。例如 Redis1 上存储 1、2、3 的 key，Redis2 上存储 4、5、6 的数据，客户端只通过一套统一的代码获取到对应的值，那么这种我们该怎么去处理这样的问题。
代码地址：
https://gitee.com/Zyred/multi_conn
- 思路分析
结合前三篇对 HashMap 原理的文章，知道了 HashMap 的存储原理是通过一个数组，然后根据 K 的 hash 值计算出 V 存储在数组哪个下标下，有了这种思路我们就可以开始创作。那么我们在来回顾一下 HashMap 原理图:
通过图中不难发现，我的思路无非就是没有链表和红黑树而已
- 准备工作
一个配置文件：保存对 Redis 服务器的配置，而且要方便解析，最终采用 properties 类型的文件
解析配置文件工具类 ：对 properties 的解析，解析成自己能理解的 JavaBean 对象，方便后续 getter 属性
保存连接、获取连接：spring 启动之初将 Redis 连接的全部创建并且将所有的操作 API 对象保存到内存中 - 思考
问题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);
}
- 部分代码
加载配置文件，配置文件名称固定，配置文件名称 multi-redis.properties
@Configuration
public class MultiRedisConnProperties {
private final String profile = "multi-redis.properties", prefix = "redis.multi.client", command = ".";
private List 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 ns = p.stringPropertyNames();
// redis.multi.client 的长度 + 1
int len = prefix.length() + 1;
// 这里能成功获取到 client 的个数
Set 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 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 类型泛型
* @return 泛型对应的实体
*/
V get (String k, Class 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 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 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 get (String k, Class 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);
}
}
- 图解 MultiRedis
6. 总结
该功能还有待完善，本文章中只是完成了我的一个突发奇想，还未真正运用在商业或者项目中，如果有雷同的需求或者有更好的思路的可以留言与我一起讨论，一起学习。