Redis - 多集群数据源配置

前言

工作时有一个项目配置了多个redis数据源，使用时出现了指定了使用副数据源，数据却依然使用了主数据源的情况。经过排查，发现配置流程较为繁琐易错，此处做一个记录。

依赖

<dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-redis</artifactId><version>1.4.1.RELEASE</version>
</dependency>

yml配置

此处设置两个不同的redis地址，可以debug时，在redisTemplate对象中查看此时使用的到底是哪一个数据源，方便排查问题。

spring:redis:jedis:pool:maxActive: 1000minIdle: 1maxWait: 5000maxIdle: 5timeout: 6000msredis-one: # 第一个redis（主）集群配置cluster:node: localhost:6379password: 123456redis-two: # 第二个redis（其他）集群配置cluster:node: 127.0.0.1:6379password: 123456

redis多集群数据源配置类

import com.fasterxml.jackson.annotation.JsonAutoDetect;
import com.fasterxml.jackson.annotation.PropertyAccessor;
import com.fasterxml.jackson.databind.ObjectMapper;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.boot.autoconfigure.data.redis.RedisProperties;
import org.springframework.cache.annotation.CachingConfigurerSupport;
import org.springframework.cache.annotation.EnableCaching;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Primary;
import org.springframework.core.env.Environment;
import org.springframework.data.redis.connection.RedisClusterConfiguration;
import org.springframework.data.redis.connection.RedisNode;
import org.springframework.data.redis.connection.RedisStandaloneConfiguration;
import org.springframework.data.redis.connection.jedis.JedisClientConfiguration;
import org.springframework.data.redis.connection.jedis.JedisConnectionFactory;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.serializer.Jackson2JsonRedisSerializer;
import org.springframework.data.redis.serializer.RedisSerializer;
import org.springframework.data.redis.serializer.StringRedisSerializer;
import redis.clients.jedis.JedisPoolConfig;import javax.annotation.Resource;
import java.util.HashSet;
import java.util.Set;@Configuration
@EnableCaching
@Slf4j
public class RedisDataSourceConfig extends CachingConfigurerSupport {@Resourceprivate Environment environment;@Resourceprivate RedisProperties redisProperties;/*** 主业务redis操作模板** @param factoryOne* @return*/@Bean(name = "redisOneTemplate")@Primarypublic RedisTemplate<String, Object> redisOneTemplate(@Autowired @Qualifier("factoryOne") JedisConnectionFactory factoryOne) {return getRedisTemplate(factoryOne);}/*** 副redis操作模板* 注意：初始化配置有误，导致实际查询的还是第一个redis(redis-one)配置* @param factoryTwo* @return*/@Bean(name = "redisTwoTemplate")public RedisTemplate<String, Object> redisTwoTemplate(@Autowired @Qualifier("factoryTwo") JedisConnectionFactory factoryTwo) {return getRedisTemplate(factoryTwo);}/*** 副redis操作模板（真）* 真实指向 redis-two配置* @param factoryTwoReal* @return*/@Bean(name = "redisTwoRealTemplate")public RedisTemplate<String, Object> redisTwoRealTemplate(@Autowired @Qualifier("factoryTwoReal") JedisConnectionFactory factoryTwoReal) {return getRedisTemplate(factoryTwoReal);}/*============================  集群模式配置 start  ===========================*//*** 指向redis-one配置*/@Bean("factoryOne")@Primarypublic JedisConnectionFactory factoryOne(RedisStandaloneConfiguration redisConfigOne, JedisClientConfiguration clientConfig) {return new JedisConnectionFactory(redisConfigOne, clientConfig);}/*** 真实指向redis-one配置*/@Bean("factoryTwo")public JedisConnectionFactory factoryTwo(RedisStandaloneConfiguration redisConfigTwo, JedisClientConfiguration clientConfig) {return new JedisConnectionFactory(redisConfigTwo, clientConfig);}/*** 真实指向redis-two配置*/@Bean("factoryTwoReal")public JedisConnectionFactory factoryTwoReal(@Autowired @Qualifier("redisConfigTwo") RedisStandaloneConfiguration redisConfigTwo, JedisClientConfiguration clientConfig) {return new JedisConnectionFactory(redisConfigTwo, clientConfig);}// 读取第一个redis配置@Primary	@Bean("redisConfigOne")public RedisStandaloneConfiguration redisConfigOne() {String nodeStrAry = environment.getProperty("spring.redis.redis-one.cluster.nodes");String password = environment.getProperty("spring.redis.redis-one.password");assert nodeStrAry != null;assert password != null;RedisStandaloneConfiguration redisStandaloneConfiguration = new RedisStandaloneConfiguration();String[] hostPortAry = nodeStrAry.split(":");redisStandaloneConfiguration.setHostName(hostPortAry[0]);redisStandaloneConfiguration.setPort(Integer.parseInt(hostPortAry[1]));redisStandaloneConfiguration.setPassword(password);return redisStandaloneConfiguration;}// 读取第二个redis配置@Bean("redisConfigTwo")public RedisStandaloneConfiguration redisConfigTwo() {String nodeStrAry = environment.getProperty("spring.redis.redis-two.cluster.nodes");String password = environment.getProperty("spring.redis.redis-two.password");assert nodeStrAry != null;assert password != null;RedisStandaloneConfiguration redisStandaloneConfiguration = new RedisStandaloneConfiguration();String[] hostPortAry = nodeStrAry.split(":");redisStandaloneConfiguration.setHostName(hostPortAry[0]);redisStandaloneConfiguration.setPort(Integer.parseInt(hostPortAry[1]));redisStandaloneConfiguration.setPassword(password);return redisStandaloneConfiguration;}/*================================集群模式 end===============================*/@Bean("clientConfig")public JedisClientConfiguration jedisClientConfiguration() {JedisPoolConfig poolConfig = new JedisPoolConfig();poolConfig.setMaxIdle(redisProperties.getJedis().getPool().getMaxIdle());poolConfig.setMinIdle(redisProperties.getJedis().getPool().getMinIdle());poolConfig.setMaxTotal(redisProperties.getJedis().getPool().getMaxActive());poolConfig.setMaxWaitMillis(redisProperties.getJedis().getPool().getMaxWait().toMillis());return JedisClientConfiguration.builder().connectTimeout(redisProperties.getTimeout()).usePooling().poolConfig(poolConfig).build();}private RedisClusterConfiguration getRedisClusterConfiguration(String nodeStrAry, String password) {RedisClusterConfiguration redisClusterConfiguration = new RedisClusterConfiguration();String[] serverArray = nodeStrAry.split(",");Set<RedisNode> nodes = new HashSet<>();for (String ipPort : serverArray) {String[] ipAndPort = ipPort.split(":");nodes.add(new RedisNode(ipAndPort[0].trim(), Integer.parseInt(ipAndPort[1])));}redisClusterConfiguration.setClusterNodes(nodes);redisClusterConfiguration.setPassword(password);return redisClusterConfiguration;}private RedisTemplate<String, Object> getRedisTemplate(JedisConnectionFactory factory) {RedisTemplate<String, Object> redisTemplate = new RedisTemplate<>();redisTemplate.setKeySerializer(new StringRedisSerializer());redisTemplate.setValueSerializer(jackson2JsonRedisSerializer());redisTemplate.setHashKeySerializer(new StringRedisSerializer());redisTemplate.setHashValueSerializer(jackson2JsonRedisSerializer());redisTemplate.setConnectionFactory(factory);return redisTemplate;}/*** json序列化** @return*/@Beanpublic RedisSerializer<Object> jackson2JsonRedisSerializer() {Jackson2JsonRedisSerializer<Object> serializer = new Jackson2JsonRedisSerializer<Object>(Object.class);ObjectMapper mapper = new ObjectMapper();mapper.setVisibility(PropertyAccessor.ALL, JsonAutoDetect.Visibility.ANY);mapper.enableDefaultTyping(ObjectMapper.DefaultTyping.NON_FINAL);serializer.setObjectMapper(mapper);return serializer;}}

注解：
@Primary ：优先考虑注入的类
@Qualifier ：通过控制里面的字符串来匹配类上的字符串达到控制注入的效果。

思考

factoryTwo方法与factoryTwoReal方法看起来都调用了redis-two的配置方法，为什么factoryTwo实际调用的还是redis-one的配置呢？

答：区别就在于入参@Autowired @Qualifier(“redisConfigTwo”) RedisStandaloneConfiguration redisConfigTwo，factoryTwo方法未使用@Qualifier(“redisConfigTwo”)指定redis-two的配置方法，实际上注入的是标记了@Primary的redis-one配置方法。

redis工具类

为主数据源与副数据源分别创建工具类或直接使用注解指定当前使用的数据源

import com.alibaba.fastjson.JSON;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.BoundListOperations;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.ZSetOperations;
import org.springframework.stereotype.Component;
import org.springframework.util.CollectionUtils;import javax.annotation.Resource;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.TimeUnit;/***  Redis主数据源工具类*/
@Component
public class RedisOneUtils {// 指向redis-one@Resource(name = "redisOneTemplate")private RedisTemplate<String, Object> redisTemplate;/*** 指定缓存失效时间** @param key  键* @param time 时间(秒)* @return*/public boolean expire(String key, long time) {try {if (time > 0) {redisTemplate.expire(key, time, TimeUnit.SECONDS);}return true;} catch (Exception e) {e.printStackTrace();return false;}}/*** 根据key 获取过期时间** @param key 键 不能为null* @return 时间(秒) 返回0代表为永久有效*/public long getExpire(String key) {return redisTemplate.getExpire(key, TimeUnit.SECONDS);}/*** 判断key是否存在** @param key 键* @return true 存在 false不存在*/public boolean hasKey(String key) {try {return redisTemplate.hasKey(key);} catch (Exception e) {e.printStackTrace();return false;}}/*** 删除缓存** @param key 可以传一个值 或多个*/@SuppressWarnings("unchecked")public void del(String... key) {if (key != null && key.length > 0) {if (key.length == 1) {redisTemplate.delete(key[0]);} else {redisTemplate.delete(CollectionUtils.arrayToList(key));}}}//============================String=============================/*** 普通缓存获取** @param key 键* @return 值*/public Object get(String key) {return key == null ? null : redisTemplate.opsForValue().get(key);}/*** 普通缓存放入** @param key   键* @param value 值* @return true成功 false失败*/public boolean set(String key, Object value) {try {redisTemplate.opsForValue().set(key, value);return true;} catch (Exception e) {e.printStackTrace();return false;}}/*** 普通缓存放入并设置时间** @param key   键* @param value 值* @param time  时间(秒) time要大于0 如果time小于等于0 将设置无限期* @return true成功 false 失败*/public boolean set(String key, Object value, long time) {try {if (time > 0) {redisTemplate.opsForValue().set(key, value, time, TimeUnit.SECONDS);} else {set(key, value);}return true;} catch (Exception e) {e.printStackTrace();return false;}}/*** 递增** @param key   键* @param delta 要增加几(大于0)* @return*/public long incr(String key, long delta) {if (delta < 0) {throw new RuntimeException("递增因子必须大于0");}return redisTemplate.opsForValue().increment(key, delta);}/*** 递减** @param key   键* @param delta 要减少几(小于0)* @return*/public long decr(String key, long delta) {if (delta < 0) {throw new RuntimeException("递减因子必须大于0");}return redisTemplate.opsForValue().increment(key, -delta);}//================================Map=================================/*** HashGet** @param key  键 不能为null* @param item 项 不能为null* @return 值*/public Object hget(String key, String item) {return redisTemplate.opsForHash().get(key, item);}/*** 获取hashKey对应的所有键值** @param key 键* @return 对应的多个键值*/public Map<Object, Object> hmget(String key) {return redisTemplate.opsForHash().entries(key);}/*** HashSet** @param key 键* @param map 对应多个键值* @return true 成功 false 失败*/public boolean hmset(String key, Map<String, Object> map) {try {redisTemplate.opsForHash().putAll(key, map);return true;} catch (Exception e) {e.printStackTrace();return false;}}/*** HashSet 并设置时间** @param key  键* @param map  对应多个键值* @param time 时间(秒)* @return true成功 false失败*/public boolean hmset(String key, Map<String, Object> map, long time) {try {redisTemplate.opsForHash().putAll(key, map);if (time > 0) {expire(key, time);}return true;} catch (Exception e) {e.printStackTrace();return false;}}/*** 向一张hash表中放入数据,如果不存在将创建** @param key   键* @param item  项* @param value 值* @return true 成功 false失败*/public boolean hset(String key, String item, Object value) {try {redisTemplate.opsForHash().put(key, item, value);return true;} catch (Exception e) {e.printStackTrace();return false;}}/*** 向一张hash表中放入数据,如果不存在将创建** @param key   键* @param item  项* @param value 值* @param time  时间(秒)  注意:如果已存在的hash表有时间,这里将会替换原有的时间* @return true 成功 false失败*/public boolean hset(String key, String item, Object value, long time) {try {redisTemplate.opsForHash().put(key, item, value);if (time > 0) {expire(key, time);}return true;} catch (Exception e) {e.printStackTrace();return false;}}/*** 删除hash表中的值** @param key  键 不能为null* @param item 项 可以使多个 不能为null*/public void hdel(String key, Object... item) {redisTemplate.opsForHash().delete(key, item);}/*** 判断hash表中是否有该项的值** @param key  键 不能为null* @param item 项 不能为null* @return true 存在 false不存在*/public boolean hHasKey(String key, String item) {return redisTemplate.opsForHash().hasKey(key, item);}/*** hash递增 如果不存在,就会创建一个 并把新增后的值返回** @param key  键* @param item 项* @param by   要增加几(大于0)* @return*/public double hincr(String key, String item, double by) {return redisTemplate.opsForHash().increment(key, item, by);}/*** hash递减** @param key  键* @param item 项* @param by   要减少记(小于0)* @return*/public double hdecr(String key, String item, double by) {return redisTemplate.opsForHash().increment(key, item, -by);}//============================set=============================/*** 根据key获取Set中的所有值** @param key 键* @return*/public Set<Object> sGet(String key) {try {return redisTemplate.opsForSet().members(key);} catch (Exception e) {e.printStackTrace();return null;}}/*** 根据value从一个set中查询,是否存在** @param key   键* @param value 值* @return true 存在 false不存在*/public boolean sHasKey(String key, Object value) {try {return redisTemplate.opsForSet().isMember(key, value);} catch (Exception e) {e.printStackTrace();return false;}}/*** 将数据放入set缓存** @param key    键* @param values 值 可以是多个* @return 成功个数*/public long sSet(String key, Object... values) {try {return redisTemplate.opsForSet().add(key, values);} catch (Exception e) {e.printStackTrace();return 0;}}/*** 将set数据放入缓存** @param key    键* @param time   时间(秒)* @param values 值 可以是多个* @return 成功个数*/public long sSetAndTime(String key, long time, Object... values) {try {Long count = redisTemplate.opsForSet().add(key, values);if (time > 0) {expire(key, time);}return count;} catch (Exception e) {e.printStackTrace();return 0;}}/*** 获取set缓存的长度** @param key 键* @return*/public long sGetSetSize(String key) {try {return redisTemplate.opsForSet().size(key);} catch (Exception e) {e.printStackTrace();return 0;}}/*** 移除值为value的** @param key    键* @param values 值 可以是多个* @return 移除的个数*/public long setRemove(String key, Object... values) {try {Long count = redisTemplate.opsForSet().remove(key, values);return count;} catch (Exception e) {e.printStackTrace();return 0;}}//===============================list=================================/*** 获取list缓存的内容** @param key   键* @param start 开始* @param end   结束  0 到 -1代表所有值* @return*/public List<Object> lGet(String key, long start, long end) {try {return redisTemplate.opsForList().range(key, start, end);} catch (Exception e) {e.printStackTrace();return null;}}/*** 获取list缓存的长度** @param key 键* @return*/public long lGetListSize(String key) {try {return redisTemplate.opsForList().size(key);} catch (Exception e) {e.printStackTrace();return 0;}}/*** 通过索引 获取list中的值** @param key   键* @param index 索引  index>=0时， 0 表头，1 第二个元素，依次类推；index<0时，-1，表尾，-2倒数第二个元素，依次类推* @return*/public Object lGetIndex(String key, long index) {try {return redisTemplate.opsForList().index(key, index);} catch (Exception e) {e.printStackTrace();return null;}}/*** 将list放入缓存** @param key   键* @param value 值* @return*/public boolean lSet(String key, Object value) {try {redisTemplate.opsForList().rightPush(key, value);return true;} catch (Exception e) {e.printStackTrace();return false;}}/*** 将list放入缓存** @param key   键* @param value 值* @param time  时间(秒)* @return*/public boolean lSet(String key, Object value, long time) {try {redisTemplate.opsForList().rightPush(key, value);if (time > 0) {expire(key, time);}return true;} catch (Exception e) {e.printStackTrace();return false;}}/*** 将list放入缓存** @param key   键* @param value 值* @return*/public boolean lSet(String key, List<Object> value) {try {redisTemplate.opsForList().rightPushAll(key, value);return true;} catch (Exception e) {e.printStackTrace();return false;}}/*** 将list放入缓存** @param key   键* @param value 值* @param time  时间(秒)* @return*/public boolean lSet(String key, List<Object> value, long time) {try {redisTemplate.opsForList().rightPushAll(key, value);if (time > 0) {expire(key, time);}return true;} catch (Exception e) {e.printStackTrace();return false;}}/*** 根据索引修改list中的某条数据** @param key   键* @param index 索引* @param value 值* @return*/public boolean lUpdateIndex(String key, long index, Object value) {try {redisTemplate.opsForList().set(key, index, value);return true;} catch (Exception e) {e.printStackTrace();return false;}}/*** 移除N个值为value** @param key   键* @param count 移除多少个* @param value 值* @return 移除的个数*/public long lRemove(String key, long count, Object value) {try {Long remove = redisTemplate.opsForList().remove(key, count, value);return remove;} catch (Exception e) {e.printStackTrace();return 0;}}/*** 模糊查询获取key值** @param pattern* @return*/public Set keys(String pattern) {return redisTemplate.keys(pattern);}/*** 使用Redis的消息队列** @param channel* @param message 消息内容*/public void convertAndSend(String channel, Object message) {redisTemplate.convertAndSend(channel, message);}/*** 根据起始结束序号遍历Redis中的list** @param listKey* @param start   起始序号* @param end     结束序号* @return*/public List<Object> rangeList(String listKey, long start, long end) {//绑定操作BoundListOperations<String, Object> boundValueOperations = redisTemplate.boundListOps(listKey);//查询数据return boundValueOperations.range(start, end);}/*** 弹出右边的值 --- 并且移除这个值** @param listKey*/public Object rifhtPop(String listKey) {//绑定操作BoundListOperations<String, Object> boundValueOperations = redisTemplate.boundListOps(listKey);return boundValueOperations.rightPop();}/*** 有序集合添加** @param key* @param value* @param scoure*/public boolean zAdd(String key, Object value, double scoure) {ZSetOperations<String, Object> zset = redisTemplate.opsForZSet();return zset.add(key, value, scoure);}/*** 移除集合中时间过期的** @param key* @return*/public boolean removeSet(String key, long max) {try {long count = redisTemplate.opsForZSet().removeRangeByScore(key, 0, max);System.out.println("count===>" + count);if (count > 0) {return true;}} catch (Exception e) {return false;}return false;}/*** 移除集合中某个成员** @param key* @param value* @return*/public Boolean delSetNum(String key, String value) {try {long cnt = redisTemplate.opsForZSet().remove(key, value);if (cnt > 0) {return true;}} catch (Exception e) {return false;}return false;}public <T> boolean setString(String key, T value) {try {//任意类型转换成StringString val = beanToString(value);if (val == null || val.length() <= 0) {return false;}redisTemplate.opsForValue().set(key, val);return true;} catch (Exception e) {return false;}}public <T> boolean setStringTime(String key, T value, long timeout) {try {//任意类型转换成StringString val = beanToString(value);if (val == null || val.length() <= 0) {return false;}redisTemplate.opsForValue().set(key, val, timeout, TimeUnit.MILLISECONDS);return true;} catch (Exception e) {return false;}}public <T> T get(String key, Class<T> clazz) {try {Object object = redisTemplate.opsForValue().get(key);if (object == null) {return null;}String value = String.valueOf(object);return stringToBean(value, clazz);} catch (Exception e) {return null;}}public <T> boolean delete(String key) {try {if (key == null || key.length() <= 0) {return false;}redisTemplate.delete(key);return true;} catch (Exception e) {return false;}}public boolean exists(String key) {try {return redisTemplate.hasKey(key);} catch (Exception e) {return false;}}@SuppressWarnings("unchecked")private <T> T stringToBean(String value, Class<T> clazz) {if (value == null || value.length() <= 0 || clazz == null) {return null;}if (clazz == int.class || clazz == Integer.class) {return (T) Integer.valueOf(value);} else if (clazz == long.class || clazz == Long.class) {return (T) Long.valueOf(value);} else if (clazz == String.class) {return (T) value;} else {return JSON.toJavaObject(JSON.parseObject(value), clazz);}}/*** @param value 任意类型* @return String*/private <T> String beanToString(T value) {if (value == null) {return null;}Class<?> clazz = value.getClass();if (clazz == int.class || clazz == Integer.class) {return "" + value;} else if (clazz == long.class || clazz == Long.class) {return "" + value;} else if (clazz == String.class) {return (String) value;} else {return JSON.toJSONString(value);}}//=========BoundListOperations 用法 End============}