【实践篇】4.13 SpringBoot Redis 多数据源集成支持哨兵模式和Cluster集群 - ToB企服应用市场:ToB评测及商务社交产业平台

<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-json</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-configuration-processor</artifactId>
<optional>true</optional>
</dependency>
<dependency>
<groupId>org.apache.commons</groupId>
<artifactId>commons-lang3</artifactId>
</dependency>
<dependency>
<groupId>com.alibaba</groupId>
<artifactId>transmittable-thread-local</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-data-redis</artifactId>
</dependency>
<dependency>
<groupId>org.apache.commons</groupId>
<artifactId>commons-pool2</artifactId>
</dependency>

复制代码

# 示例
custom.primary.redis.key=user

复制代码

spring.redis.xxx.timeout = 3000
spring.redis.xxx.maxTotal = 50
spring.redis.xxx.maxIdle = 50
spring.redis.xxx.minIdle = 2
spring.redis.xxx.maxWaitMillis = 10000
spring.redis.xxx.testOnBorrow = False

复制代码

# 第1个Redis 实例用于用户体系,我们取标识为user
spring.redis.user.host = 127.0.0.1
spring.redis.user.port = 6380
spring.redis.user.password = 密码
spring.redis.user.database = 0
# 第2个Redis 实例用于IoT体系
spring.redis.iot.host = 127.0.0.1
spring.redis.iot.port = 6390
spring.redis.iot.password = 密码
spring.redis.iot.database = 0
# 第3个Redis 实例用于xxx
spring.redis.xxx.host = 127.0.0.1
spring.redis.xxx.port = 6390
spring.redis.xxx.password = 密码
spring.redis.xxx.database = 0

复制代码

spring.redis.xxx1.sentinel.master=mymaster1
spring.redis.xxx1.sentinel.nodes=ip:端口，ip:端口
spring.redis.xxx1.password = bD945aAfeb422E22AbAdFb9D2a22bEDd
spring.redis.xxx1.database = 0
spring.redis.xxx1.timeout = 3000
#第二个
spring.redis.xxx2.sentinel.master=mymaster2
spring.redis.xxx2.sentinel.nodes=ip:端口,ip:端口
spring.redis.xxx2.password = bD945aAfeb422E22AbAdFb9D2a22bEDd
spring.redis.xxx2.database = 0
spring.redis.xxx2.timeout = 3000

复制代码

spring.redis.xxx1.cluster.nodes=ip1:端口,ip2:端口,ip3:端口,ip4:端口,ip5:端口,ip6:端口
spring.redis.xxx1.cluster.max-redirects=5
spring.redis.xxx1.password = 密码
spring.redis.xxx1.timeout = 3000

复制代码

// 定义静态Map变量redis，用于存储Redis配置参数
protected static Map<String, Map<String, String>> redis = new HashMap<>();

复制代码

private RedisStandaloneConfiguration buildStandaloneConfig(Map<String, String> param){
//...省略
}

复制代码

private RedisSentinelConfiguration buildSentinelConfig(Map<String, String> param){
//...省略
}

复制代码

private RedisClusterConfiguration buildClusterConfig(Map<String, String> param){
//...省略
}

复制代码

public RedisConnectionFactory buildLettuceConnectionFactory(String redisKey, Map<String, String> param,GenericObjectPoolConfig genericObjectPoolConfig){
...
}

复制代码

package com.iceicepip.project.common.redis;import com.iceicepip.project.common.redis.util.AddressUtils;import com.fasterxml.jackson.annotation.JsonAutoDetect;import com.fasterxml.jackson.annotation.PropertyAccessor;import com.fasterxml.jackson.databind.ObjectMapper;import org.apache.commons.lang3.StringUtils;import org.apache.commons.pool2.impl.GenericObjectPoolConfig;import org.springframework.beans.MutablePropertyValues;import org.springframework.beans.factory.InitializingBean;import org.springframework.beans.factory.annotation.Value;import org.springframework.beans.factory.config.BeanPostProcessor;import org.springframework.beans.factory.config.ConstructorArgumentValues;import org.springframework.beans.factory.support.DefaultListableBeanFactory;import org.springframework.beans.factory.support.GenericBeanDefinition;import org.springframework.boot.autoconfigure.AutoConfiguration;import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;import org.springframework.boot.context.properties.ConfigurationProperties;import org.springframework.context.ApplicationContext;import org.springframework.context.ApplicationContextAware;import org.springframework.context.annotation.Bean;import org.springframework.core.env.MapPropertySource;import org.springframework.core.env.StandardEnvironment;import org.springframework.data.redis.connection.*;import org.springframework.data.redis.connection.lettuce.LettuceClientConfiguration;import org.springframework.data.redis.connection.lettuce.LettuceConnectionFactory;import org.springframework.data.redis.connection.lettuce.LettucePoolingClientConfiguration;import org.springframework.data.redis.core.RedisTemplate;import org.springframework.data.redis.core.StringRedisTemplate;import org.springframework.data.redis.serializer.Jackson2JsonRedisSerializer;import org.springframework.data.redis.serializer.StringRedisSerializer;import java.time.Duration;import java.util.*;@AutoConfiguration@ConfigurationProperties(prefix = "spring")public class CustomRedisConfig implements InitializingBean, ApplicationContextAware, BeanPostProcessor { // 定义静态Map变量redis，用于存储Redis配置参数
protected static Map<String, Map<String, String>> redis = new HashMap<>();
// 在代码中作为Redis的主数据源的标识 @Value("${customer.primary.redis.key}") private String primaryKey; @Override // 实现InitializingBean接口的方法，用于在属性被注入后初始化Redis毗连工厂和Redis模板 public void afterPropertiesSet() { redis.forEach((k, v) -> { // 如果当前的Redis主键等于注入的主键，则将Redis配置参数参加到属性源中 if(Objects.equals(k,primaryKey)){ Map<String, Object> paramMap = new HashMap<>(4); v.forEach((k1,v1)-> paramMap.put("spring.redis."+k1,v1)); MapPropertySource mapPropertySource = new MapPropertySource("redisAutoConfigProperty", paramMap); ((StandardEnvironment)applicationContext.getEnvironment()).getPropertySources().addLast(mapPropertySource); } // 创建Redis毗连池配置对象和毗连工厂对象 GenericObjectPoolConfig genericObjectPoolConfig = buildGenericObjectPoolConfig(k, v); RedisConnectionFactory lettuceConnectionFactory = buildLettuceConnectionFactory(k, v, genericObjectPoolConfig); // 创建Redis模板对象和字符串模板对象，并调用方法创建自定义Redis服务对象 RedisTemplate redisTemplate = buildRedisTemplate(k, lettuceConnectionFactory); StringRedisTemplate stringRedisTemplate = buildStringRedisTemplate(k, lettuceConnectionFactory); buildCustomRedisService(k, redisTemplate, stringRedisTemplate); }); } // 创建Redis主数据源 RedisTemplate @Bean public RedisTemplate<Object, Object> redisTemplate() { Map<String, String> redisParam = redis.get(primaryKey); GenericObjectPoolConfig genericObjectPoolConfig = buildGenericObjectPoolConfig(primaryKey, redisParam); RedisConnectionFactory lettuceConnectionFactory = buildLettuceConnectionFactory(primaryKey, redisParam, genericObjectPoolConfig); RedisTemplate<Object, Object> template = new RedisTemplate(); template.setConnectionFactory(lettuceConnectionFactory); return template; } // 创建Redis主数据源 StringRedisTemplate @Bean @ConditionalOnMissingBean public StringRedisTemplate stringRedisTemplate() { Map<String, String> redisParam = redis.get(primaryKey); GenericObjectPoolConfig genericObjectPoolConfig = buildGenericObjectPoolConfig(primaryKey, redisParam); RedisConnectionFactory lettuceConnectionFactory = buildLettuceConnectionFactory(primaryKey, redisParam, genericObjectPoolConfig); StringRedisTemplate template = new StringRedisTemplate(); template.setConnectionFactory(lettuceConnectionFactory); return template; } // 创建自定义Redis服务对象 private void buildCustomRedisService(String k, RedisTemplate redisTemplate, StringRedisTemplate stringRedisTemplate) { ConstructorArgumentValues constructorArgumentValues = new ConstructorArgumentValues(); constructorArgumentValues.addIndexedArgumentValue(0, stringRedisTemplate); constructorArgumentValues.addIndexedArgumentValue(1, redisTemplate); // 将来利用的时间Redis对象的beanName,区分多个数据源 setCosBean(k + "Redis", CustomRedisService.class, constructorArgumentValues); } // 创建StringRedisTemplate private StringRedisTemplate buildStringRedisTemplate(String k, RedisConnectionFactory lettuceConnectionFactory) { ConstructorArgumentValues constructorArgumentValues = new ConstructorArgumentValues(); constructorArgumentValues.addIndexedArgumentValue(0, lettuceConnectionFactory); setCosBean(k + "StringRedisTemplate", StringRedisTemplate.class, constructorArgumentValues); return getBean(k + "StringRedisTemplate"); } // 创建Redis模板对象 private RedisTemplate buildRedisTemplate(String k, RedisConnectionFactory lettuceConnectionFactory) { // 如果已经存在Redis模板对象，则直接返回该对象 if(applicationContext.containsBean(k + "RedisTemplate")){ return getBean(k + "RedisTemplate"); } // 创建Redis序列化器对象 Jackson2JsonRedisSerializer serializer = new Jackson2JsonRedisSerializer(Object.class); ObjectMapper mapper = new ObjectMapper(); mapper.setVisibility(PropertyAccessor.ALL, JsonAutoDetect.Visibility.ANY); mapper.enableDefaultTyping(ObjectMapper.DefaultTyping.NON_FINAL); serializer.setObjectMapper(mapper); // 创建Redis模板对象，并设置毗连池配置和序列化器等属性 Map original = new HashMap<>(2); original.put("connectionFactory", lettuceConnectionFactory); original.put("valueSerializer", serializer); original.put("keySerializer", new StringRedisSerializer()); original.put("hashKeySerializer", new StringRedisSerializer()); original.put("hashValueSerializer", serializer); // 将来利用RedisTemplate的地方只需要用注解制定beanName 即可获取到每个Redis实例的利用工具类 setBean(k + "RedisTemplate", RedisTemplate.class, original); return getBean(k + "RedisTemplate"); }} public GenericObjectPoolConfig buildGenericObjectPoolConfig(String redisKey, Map<String, String> param) { if(applicationContext.containsBean(redisKey + "GenericObjectPoolConfig")){ return getBean(redisKey + "GenericObjectPoolConfig"); } Integer maxIdle = StringUtils.isEmpty(param.get(CustomRedisConfigConstant.REDIS_MAXIDLE)) ? GenericObjectPoolConfig.DEFAULT_MAX_IDLE : Integer.valueOf(param.get(CustomRedisConfigConstant.REDIS_MAXIDLE)); Integer minIdle = StringUtils.isEmpty(param.get(CustomRedisConfigConstant.REDIS_MINIDLE)) ? GenericObjectPoolConfig.DEFAULT_MIN_IDLE : Integer.valueOf(param.get(CustomRedisConfigConstant.REDIS_MINIDLE)); Integer maxTotal = StringUtils.isEmpty(param.get(CustomRedisConfigConstant.REDIS_MAXTOTAL)) ? GenericObjectPoolConfig.DEFAULT_MAX_TOTAL : Integer.valueOf(param.get(CustomRedisConfigConstant.REDIS_MAXTOTAL)); Long maxWaitMillis = StringUtils.isEmpty(param.get(CustomRedisConfigConstant.REDIS_MAXWAITMILLIS)) ? -1L:Long.valueOf(param.get(CustomRedisConfigConstant.REDIS_MAXWAITMILLIS)); Boolean testOnBorrow = StringUtils.isEmpty(param.get(CustomRedisConfigConstant.REDIS_TESTONBORROW)) ? Boolean.FALSE :Boolean.valueOf(param.get(CustomRedisConfigConstant.REDIS_TESTONBORROW)); Map original = new HashMap<>(8); original.put("maxTotal", maxTotal); original.put("maxIdle", maxIdle); original.put("minIdle", minIdle); original.put("maxWaitMillis",maxWaitMillis); original.put("testOnBorrow",testOnBorrow); setBean(redisKey + "GenericObjectPoolConfig", GenericObjectPoolConfig.class, original); return getBean(redisKey + "GenericObjectPoolConfig"); } public RedisConnectionFactory buildLettuceConnectionFactory(String redisKey, Map<String, String> param,GenericObjectPoolConfig genericObjectPoolConfig){ if(applicationContext.containsBean(redisKey + "redisConnectionFactory")){ return getBean(redisKey + "redisConnectionFactory"); } String timeout = StringUtils.defaultIfEmpty(param.get(CustomRedisConfigConstant.REDIS_TIMEOUT), "3000"); LettuceClientConfiguration clientConfig = LettucePoolingClientConfiguration.builder() .commandTimeout(Duration.ofMillis(Long.valueOf(timeout))) .poolConfig(genericObjectPoolConfig) .build(); Object firstArgument = null; if(this.isCluster(param)){ RedisClusterConfiguration clusterConfiguration = buildClusterConfig(param); firstArgument = clusterConfiguration; } else if(this.isSentinel(param)){ RedisSentinelConfiguration sentinelConfiguration = buildSentinelConfig(param); firstArgument = sentinelConfiguration; } else{ RedisStandaloneConfiguration standaloneConfiguration = buildStandaloneConfig(param); firstArgument = standaloneConfiguration; } ConstructorArgumentValues constructorArgumentValues = new ConstructorArgumentValues(); constructorArgumentValues.addIndexedArgumentValue(0, firstArgument); constructorArgumentValues.addIndexedArgumentValue(1, clientConfig); setCosBean(redisKey + "redisConnectionFactory", LettuceConnectionFactory.class, constructorArgumentValues); return getBean(redisKey +"redisConnectionFactory"); } /** * 如果配置的是哨兵模式 * @return */ private boolean isSentinel(Map<String, String> param){ String sentinelMaster = param.get(CustomRedisConfigConstant.REDIS_SENTINEL_MASTER); String sentinelNodes = param.get(CustomRedisConfigConstant.REDIS_SENTINEL_NODES); return StringUtils.isNotEmpty(sentinelMaster) && StringUtils.isNotEmpty(sentinelNodes); } /** * 如果配置的是集群模式 * @return */ private boolean isCluster(Map<String, String> param){ String clusterNodes = param.get(CustomRedisConfigConstant.REDIS_CLUSTER_NODES); return StringUtils.isNotEmpty(clusterNodes); } private RedisStandaloneConfiguration buildStandaloneConfig(Map<String, String> param){ String host = param.get(CustomRedisConfigConstant.REDIS_HOST); String port = param.get(CustomRedisConfigConstant.REDIS_PORT); String database = param.get(CustomRedisConfigConstant.REDIS_DATABASE); String password = param.get(CustomRedisConfigConstant.REDIS_PASSWORD); RedisStandaloneConfiguration standaloneConfig = new RedisStandaloneConfiguration(); standaloneConfig.setHostName(host); standaloneConfig.setDatabase(Integer.valueOf(database)); standaloneConfig.setPort(Integer.valueOf(port)); standaloneConfig.setPassword(RedisPassword.of(password)); return standaloneConfig; } private RedisSentinelConfiguration buildSentinelConfig(Map<String, String> param){ String sentinelMaster = param.get(CustomRedisConfigConstant.REDIS_SENTINEL_MASTER); String sentinelNodes = param.get(CustomRedisConfigConstant.REDIS_SENTINEL_NODES); String database = param.get(CustomRedisConfigConstant.REDIS_DATABASE); String password = param.get(CustomRedisConfigConstant.REDIS_PASSWORD); RedisSentinelConfiguration config = new RedisSentinelConfiguration(); config.setMaster(sentinelMaster); Iterable<AddressUtils.Address> addressIterable = AddressUtils.parseAddresses(sentinelNodes); Iterable<RedisNode> redisNodes = transform(addressIterable); config.setDatabase(Integer.valueOf(database)); config.setPassword(RedisPassword.of(password)); config.setSentinels(redisNodes); return config; } private RedisClusterConfiguration buildClusterConfig(Map<String, String> param){ String clusterNodes = param.get(CustomRedisConfigConstant.REDIS_CLUSTER_NODES); String clusterMaxRedirects = param.get(CustomRedisConfigConstant.REDIS_CLUSTER_MAX_REDIRECTS); String password = param.get(CustomRedisConfigConstant.REDIS_PASSWORD); RedisClusterConfiguration config = new RedisClusterConfiguration(); Iterable<AddressUtils.Address> addressIterable = AddressUtils.parseAddresses(clusterNodes); Iterable<RedisNode> redisNodes = transform(addressIterable); config.setClusterNodes(redisNodes); config.setMaxRedirects(StringUtils.isEmpty(clusterMaxRedirects) ? 5 : Integer.valueOf(clusterMaxRedirects)); config.setPassword(RedisPassword.of(password)); return config; } private Iterable<RedisNode> transform(Iterable<AddressUtils.Address> addresses){ List<RedisNode> redisNodes = new ArrayList<>(); addresses.forEach( address -> redisNodes.add(new RedisServer(address.getHost(), address.getPort()))); return redisNodes; } private static ApplicationContext applicationContext; public Map<String, Map<String, String>> getRedis() { return redis; } /** * 实现ApplicationContextAware接口的context注入函数, 将其存入静态变量. */ @Override public void setApplicationContext(ApplicationContext applicationContext) { CustomRedisConfig.applicationContext = applicationContext; } private static void checkApplicationContext() { if (applicationContext == null) { throw new IllegalStateException("applicaitonContext未注入,请在applicationContext.xml中定义SpringContextUtil"); } } /** * 从静态变量ApplicationContext中取得Bean, 主动转型为所赋值对象的类型. */ @SuppressWarnings("unchecked") public static <T> T getBean(String name) { checkApplicationContext(); if (applicationContext.containsBean(name)) { return (T) applicationContext.getBean(name); } return null; } /** * 删除spring中管理的bean * * @param beanName */ public static void removeBean(String beanName) { DefaultListableBeanFactory acf = (DefaultListableBeanFactory) applicationContext.getAutowireCapableBeanFactory(); acf.removeBeanDefinition(beanName); } /** * 同步方法注册bean到ApplicationContext中 * * @param beanName * @param clazz * @param original bean的属性值 */ public synchronized void setBean(String beanName, Class<?> clazz, Map<String, Object> original) { checkApplicationContext(); DefaultListableBeanFactory beanFactory = (DefaultListableBeanFactory) applicationContext.getAutowireCapableBeanFactory(); if (beanFactory.containsBean(beanName)) { return; } GenericBeanDefinition definition = new GenericBeanDefinition(); //类class definition.setBeanClass(clazz); if(beanName.startsWith(primaryKey)){ definition.setPrimary(true); } //属性赋值 definition.setPropertyValues(new MutablePropertyValues(original)); //注册到spring上下文 beanFactory.registerBeanDefinition(beanName, definition); } public synchronized void setCosBean(String beanName, Class<?> clazz, ConstructorArgumentValues original) { checkApplicationContext(); DefaultListableBeanFactory beanFactory = (DefaultListableBeanFactory) applicationContext.getAutowireCapableBeanFactory(); //这里紧张 if (beanFactory.containsBean(beanName)) { return; } GenericBeanDefinition definition = new GenericBeanDefinition(); //类class definition.setBeanClass(clazz); if(beanName.startsWith(primaryKey)){ definition.setPrimary(true); } //属性赋值 definition.setConstructorArgumentValues(new ConstructorArgumentValues(original)); //注册到spring上下文 beanFactory.registerBeanDefinition(beanName, definition); }}

复制代码

package com.iceicepip.project.common.redis;
public class CustomRedisConfigConstant {
private CustomRedisConfigConstant() {
}
public static final String REDIS_HOST = "host";
public static final String REDIS_PORT = "port";
public static final String REDIS_TIMEOUT = "timeout";
public static final String REDIS_DATABASE = "database";
public static final String REDIS_PASSWORD = "password";
public static final String REDIS_MAXWAITMILLIS = "maxWaitMillis";
public static final String REDIS_MAXIDLE = "maxIdle";
public static final String REDIS_MINIDLE = "minIdle";
public static final String REDIS_MAXTOTAL = "maxTotal";
public static final String REDIS_TESTONBORROW = "testOnBorrow";
public static final String REDIS_SENTINEL_MASTER = "sentinel.master";
public static final String REDIS_SENTINEL_NODES = "sentinel.nodes";
public static final String REDIS_CLUSTER_NODES = "cluster.nodes";
public static final String REDIS_CLUSTER_MAX_REDIRECTS = "cluster.max-redirects";
public static final String BEAN_NAME_SUFFIX = "Redis";
public static final String INIT_METHOD_NAME = "getInit";
}

复制代码

package com.iceicepip.project.common.redis;
import com.alibaba.ttl.TransmittableThreadLocal;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.aop.framework.Advised;
import org.springframework.aop.support.AopUtils;
import org.springframework.beans.factory.NoSuchBeanDefinitionException;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.ApplicationContext;
import org.springframework.dao.DataAccessException;
import org.springframework.data.redis.connection.RedisConnection;
import org.springframework.data.redis.connection.RedisStringCommands;
import org.springframework.data.redis.connection.ReturnType;
import org.springframework.data.redis.connection.StringRedisConnection;
import org.springframework.data.redis.core.*;
import org.springframework.data.redis.core.types.Expiration;
import javax.annotation.Resource;
import java.nio.charset.StandardCharsets;
import java.util.*;
import java.util.concurrent.TimeUnit;
import java.util.function.Function;
import java.util.function.Supplier;
public class CustomRedisService {
private static final Logger logger = LoggerFactory.getLogger(CustomRedisService.class);
private StringRedisTemplate stringRedisTemplate;
private RedisTemplate redisTemplate;
@Value("${distribute.lock.MaxSeconds:100}")
private Integer lockMaxSeconds;
private static Long LOCK_WAIT_MAX_TIME = 120000L;
@Resource
private ApplicationContext applicationContext;
/**
* 保存锁的value
*/
private TransmittableThreadLocal<String> redisLockReentrant = new TransmittableThreadLocal<>();
/**
* 解锁lua脚本
*/
private static final String RELEASE_LOCK_SCRIPT = "if redis.call('get', KEYS[1]) == ARGV[1] then return redis.call('del', KEYS[1]) else return 0 end";
/**
* redis锁固定前缀
*/
private static final String REDIS_LOCK_KEY_PREFIX = "xxx:redisLock";
/**
* redis nameSpace
*/
private static final String REDIS_NAMESPACE_PREFIX = ":";
@Value("${spring.application.name}")
private String appName;
public CustomRedisService() {
}
public CustomRedisService(StringRedisTemplate stringRedisTemplate, RedisTemplate redisTemplate) {
this.stringRedisTemplate = stringRedisTemplate;
this.redisTemplate = redisTemplate;
}
public StringRedisTemplate getStringRedisTemplate() {
return stringRedisTemplate;
}
public RedisTemplate getRedisTemplate() {
return redisTemplate;
}
//以下是操作
public void saveOrUpdate(HashMap<String, String> values) throws Exception {
ValueOperations<String, String> valueOps = stringRedisTemplate
.opsForValue();
valueOps.multiSet(values);
}
public void saveOrUpdate(String key, String value) throws Exception {
ValueOperations<String, String> valueOps = stringRedisTemplate
.opsForValue();
valueOps.set(key, value);
}
public String getValue(String key) throws Exception {
ValueOperations<String, String> valueOps = stringRedisTemplate
.opsForValue();
return valueOps.get(key);
}
public void setValue(String key, String value) throws Exception {
ValueOperations<String, String> valueOps = stringRedisTemplate
.opsForValue();
valueOps.set(key, value);
}
public void setValue(String key, String value, long timeout, TimeUnit unit) throws Exception {
ValueOperations<String, String> valueOps = stringRedisTemplate
.opsForValue();
valueOps.set(key, value, timeout, unit);
}
public List<String> getValues(Collection<String> keys) throws Exception {
ValueOperations<String, String> valueOps = stringRedisTemplate
.opsForValue();
return valueOps.multiGet(keys);
}
public void delete(String key) throws Exception {
stringRedisTemplate.delete(key);
}
public void delete(Collection<String> keys) throws Exception {
stringRedisTemplate.delete(keys);
}
public void addSetValues(String key, String... values) throws Exception {
SetOperations<String, String> setOps = stringRedisTemplate.opsForSet();
setOps.add(key, values);
}
public Set<String> getSetValues(String key) throws Exception {
SetOperations<String, String> setOps = stringRedisTemplate.opsForSet();
return setOps.members(key);
}
public String getSetRandomMember(String key) throws Exception {
SetOperations<String, String> setOps = stringRedisTemplate.opsForSet();
return setOps.randomMember(key);
}
public void delSetValues(String key, Object... values) throws Exception {
SetOperations<String, String> setOps = stringRedisTemplate.opsForSet();
setOps.remove(key, values);
}
public Long getZsetValuesCount(String key) throws Exception {
return stringRedisTemplate.opsForSet().size(key);
}
public void addHashSet(String key, HashMap<String, String> args)
throws Exception {
HashOperations<String, String, String> hashsetOps = stringRedisTemplate
.opsForHash();
hashsetOps.putAll(key, args);
}
public Map<String, String> getHashSet(String key) throws Exception {
HashOperations<String, String, String> hashsetOps = stringRedisTemplate
.opsForHash();
return hashsetOps.entries(key);
}
public Map<byte[], byte[]> getHashByteSet(String key) throws Exception {
RedisConnection connection = null;
try {
connection = redisTemplate.getConnectionFactory().getConnection();
return connection.hGetAll(key.getBytes());
} catch (Exception e) {
throw new Exception(e);
} finally {
if (Objects.nonNull(connection) && !connection.isClosed()) {
connection.close();
}
}
}
public List<byte[]> getHashMSet(byte[] key, byte[][] fields) throws Exception {
return stringRedisTemplate.getConnectionFactory().getConnection().hMGet(key, fields);
}
/**
* 设备hash中的值
*
* @param key
* @param field
* @param vaule
* @return
* @throws Exception
*/
public Boolean setHashMSet(byte[] key, byte[] field, byte[] vaule) throws Exception {
return stringRedisTemplate.getConnectionFactory().getConnection().hSet(key, field, vaule);
}
/**
* 采用Pipeline方式获取多个Key的数据
*
* @param keys Key数组
* @param fields Hash对象的二级Key
* @return 结果数组，每个Object对象为List<byte[]>，使用时需判断是否为null
* @throws Exception
*/
public List<Object> getHashMSet(byte[][] keys, byte[][] fields) throws Exception {
if (keys == null || keys.length == 0 || fields == null || fields.length == 0) {
return null;
}
RedisConnection connection = stringRedisTemplate.getConnectionFactory().getConnection();
try {
connection.openPipeline();
for (byte[] key : keys) {
connection.hMGet(key, fields);
}
return connection.closePipeline();
} finally {
if (!connection.isClosed()) {
connection.close();
}
}
}
/**
* 采用Pipeline方式获取多个Key的数据
*
* @param keys Key数组
* @param field Hash对象的二级Key
* @return 结果数组，每个Object对象为List<byte[]>，使用时需判断是否为null
* @throws Exception
*/
public List<Object> getHashMSet(byte[][] keys, byte[] field) throws Exception {
if (keys == null || keys.length == 0 || field == null) {
return null;
}
RedisConnection connection = stringRedisTemplate.getConnectionFactory().getConnection();
try {
connection.openPipeline();
for (byte[] key : keys) {
connection.hGet(key, field);
}
return connection.closePipeline();
} finally {
if (!connection.isClosed()) {
connection.close();
}
}
}
/**
* 采用Pipeline方式获取多个Key的数据
*
* @param keys Key数组
* @return 结果数组，每个Object对象为List<byte[]>，使用时需判断是否为null
* @throws Exception
*/
public List<Object> getHashMSet(byte[][] keys) throws Exception {
if (keys == null || keys.length == 0) {
return null;
}
RedisConnection connection = stringRedisTemplate.getConnectionFactory().getConnection();
try {
connection.openPipeline();
for (byte[] key : keys) {
connection.hGetAll(key);
}
return connection.closePipeline();
} finally {
if (!connection.isClosed()) {
connection.close();
}
}
}
/**
* 删除批量string
*
* @param keys Key数组
*/
public void deleteAllStringValues(byte[][] keys) {
if (keys == null || keys.length == 0) {
return;
}
RedisConnection connection = stringRedisTemplate.getConnectionFactory().getConnection();
try {
connection.openPipeline();
for (byte[] key : keys) {
connection.del(key);
}
connection.closePipeline();
} finally {
if (!connection.isClosed()) {
connection.close();
}
}
}
public List<String> getHashMSet(String key, List<String> fields) throws Exception {
HashOperations<String, String, String> hashsetOps = stringRedisTemplate
.opsForHash();
return hashsetOps.multiGet(key, fields);
}
public List<byte[]> getHashByteMSet(String key, byte[]... fields) throws Exception {
// HashOperations<String, String, byte[]> hashsetOps = stringRedisTemplate
// .opsForHash();
// return hashsetOps.multiGet(key, fields);
RedisConnection connection = null;
try {
connection = redisTemplate.getConnectionFactory().getConnection();
return connection.hMGet(key.getBytes(), fields);
} catch (Exception e) {
throw new Exception(e);
} finally {
if (Objects.nonNull(connection) && !connection.isClosed()) {
connection.close();
}
}
}
public void delHashSetValues(String key, Object... values) throws Exception {
HashOperations<String, String, String> hashsetOps = stringRedisTemplate
.opsForHash();
hashsetOps.delete(key, values);
}
public void addZset(String key, String value, double score)
throws Exception {
ZSetOperations<String, String> zSetOps = stringRedisTemplate
.opsForZSet();
zSetOps.add(key, value, score);
}
public Set<String> getZsetValues(String key) throws Exception {
return null;
}
public void delZsetValues(String key, Object... values) throws Exception {
ZSetOperations<String, String> zSetOps = stringRedisTemplate
.opsForZSet();
zSetOps.remove(key, values);
}
public String getHashByKey(String redisKey, String mapKey) throws Exception {
HashOperations<String, String, String> hashsetOps = stringRedisTemplate.opsForHash();
return hashsetOps.get(redisKey, mapKey);
}
public byte[] getHashByteByKey(String redisKey, String mapKey) throws Exception {
RedisConnection connection = null;
try {
connection = redisTemplate.getConnectionFactory().getConnection();
return connection.hGet(redisKey.getBytes(), mapKey.getBytes());
} catch (Exception e) {
throw new Exception(e);
} finally {
if (Objects.nonNull(connection) && !connection.isClosed()) {
connection.close();
}
}
// HashOperations<String, String, byte[]> hashsetOps = stringRedisTemplate.opsForHash();
// return hashsetOps.get(redisKey, mapKey);
}
public Map<byte[], byte[]> getHashByte(String redisKey) throws Exception {
RedisConnection connection = null;
try {
connection = redisTemplate.getConnectionFactory().getConnection();
return connection.hGetAll(redisKey.getBytes());
} catch (Exception e) {
throw new Exception(e);
} finally {
if (Objects.nonNull(connection) && !connection.isClosed()) {
connection.close();
}
}
}
public void addHashSet(String redisKey, String mapKey, String mapValue)
throws Exception {
stringRedisTemplate.opsForHash().put(redisKey, mapKey, mapValue);
}
public Set<String> getSet(String key) throws Exception {
SetOperations<String, String> setOperations = stringRedisTemplate.opsForSet();
return setOperations.members(key);
}
public void addSetValuesPipelined(final String[] keys, final String value) throws Exception {
stringRedisTemplate.executePipelined(new RedisCallback<Object>() {
@Override
public Object doInRedis(RedisConnection connection) {
StringRedisConnection stringRedisConn = (StringRedisConnection) connection;
for (int i = 0; i < keys.length; i++) {
stringRedisConn.sAdd(keys[i], value);
}
//必须返回null
return null;
}
});
}
public void delSetValuesPipelined(final String[] keys, final String value) throws Exception {
stringRedisTemplate.executePipelined(new RedisCallback<Object>() {
@Override
public Object doInRedis(RedisConnection connection) {
StringRedisConnection stringRedisConn = (StringRedisConnection) connection;
for (int i = 0; i < keys.length; i++) {
stringRedisConn.sRem(keys[i], value);
}
//必须返回null
return null;
}
});
}
public void delHashByKey(String redisKey, String mapKey) throws Exception {
HashOperations<String, String, String> hashMapOps = stringRedisTemplate.opsForHash();
hashMapOps.delete(redisKey, mapKey);
}
public Boolean hasKey(String key) throws Exception {
return stringRedisTemplate.hasKey(key);
}
/**
* 设置用户其他类的缓存
*
* @param key
* @param field hash结构的field
* @param data 需要存的数据
* @param timeOut 超时时间
* @param unit 时间单位
*/
public void setHashOther(String key, String field, String data, long timeOut, TimeUnit unit) {
stringRedisTemplate.opsForHash().put(key, field, data);
stringRedisTemplate.expire(key, timeOut, unit);
}
/**
* 返回用户的其他缓存
*
* @param key
* @param field hash结构的field
* @return String
* @throws Exception
*/
public String getHashOther(String key, String field) throws Exception {
return this.getHashByKey(key, field);
}
/**
* 2019-2-20 changyandong 新增incr方法，设置过期时间
*
* @param key
* @param delta
* @param timeout
* @param unit
* @return
*/
public Long increment(final String key, final int delta, final long timeout,
final TimeUnit unit) {
if (timeout <= 0 || unit == null) {
return stringRedisTemplate.opsForValue().increment(key, delta);
}
List<Object> result = stringRedisTemplate
.executePipelined(new SessionCallback<Object>() {
@Override
public <K, V> Object execute(
RedisOperations<K, V> operations)
throws DataAccessException {
ValueOperations<K, V> ops = operations.opsForValue();
ops.increment((K) key, delta);
operations.expire((K) key, timeout, unit);
return null;
}
});
return (Long) result.get(0);
}
/**
* 管道增加hash结构
*/
public void addHashValuesPipelined(Map<String, Map<String, String>> keys) {
stringRedisTemplate.executePipelined((RedisCallback<Object>) connection -> {
StringRedisConnection stringRedisConn = (StringRedisConnection) connection;
keys.forEach(stringRedisConn::hMSet);
//必须返回null
return null;
});
}
/**
* 管道增加hash结构删除老hash结构
*/
public void addHashValuesPipelinedRemoveOldHash(Map<String, Map<String, String>> keys) {
stringRedisTemplate.executePipelined((RedisCallback<Object>) connection -> {
StringRedisConnection stringRedisConn = (StringRedisConnection) connection;
stringRedisConn.del(keys.keySet().toArray(new String[0]));
keys.forEach(stringRedisConn::hMSet);
//必须返回null
return null;
});
}
/**
* 分布式锁模板方法
*
* @param businessKey 业务key
* @param callbackFunction 回调方法
* @param s 回调方法具体入参
* @param <S> 回调方法入参类型
* @param <T> 回调方法返回值类型
* @return 回调方法返回值
*/
public <S, T> T redisLockCallback(String businessKey, Function<S, T> callbackFunction, S s) {
try {
redisLock(businessKey);
return callbackFunction.apply(s);
} finally {
redisUnLock(businessKey);
}
}
public <T> T redisLockSupplier(String businessKey, Supplier<T> supplier) {
return redisLockSupplier(businessKey, supplier, lockMaxSeconds, LOCK_WAIT_MAX_TIME, TimeUnit.SECONDS);
}
public <T> T redisLockSupplier(String businessKey, Supplier<T> supplier, long lockMaxTime, long tryTimeout, TimeUnit timeUnit) {
try {
redisLock(businessKey, lockMaxTime, tryTimeout, timeUnit);
return supplier.get();
} finally {
redisUnLock(businessKey);
}
}
/**
* 获取锁（不等待，直接返回是否获取到锁资源）
*
* @param businessKey 业务key
* @return 是否获取到锁资源
*/
public boolean redisLockSuspend(String businessKey) {
return redisLockSuspend(businessKey, lockMaxSeconds, TimeUnit.SECONDS);
}
/**
* 获取锁（不等待，直接返回是否获取到锁资源）
* @param businessKey 业务key
* @param lockMaxTime 锁占用时长
* @param timeUnit 时间单位
* @return 是否获取锁资源
*/
public boolean redisLockSuspend(String businessKey, long lockMaxTime, TimeUnit timeUnit) {
String lockKey = generateLockKey(businessKey);
long finalLockMaxTime = timeUnit.toMillis(lockMaxTime);
//可重入锁判断
if (isReentrantLock(lockKey)) {
return Boolean.TRUE;
}
RedisCallback<Boolean> callback = (connection) -> connection.set(
lockKey.getBytes(StandardCharsets.UTF_8),
businessKey.getBytes(StandardCharsets.UTF_8),
Expiration.milliseconds(finalLockMaxTime),
RedisStringCommands.SetOption.SET_IF_ABSENT);
return stringRedisTemplate.execute(callback);
}
/**
* @param keyPrefix redis锁 key前缀
* @param key key
* @param tryTimeout 超时时间
* @param timeUnit 时间单位
* @return 是否获取到锁资源
*/
@Deprecated
public boolean redisLock(String keyPrefix, String key, long lockMaxTime, long tryTimeout, TimeUnit timeUnit) {
String businessKey = getLockKey(keyPrefix, key);
return redisLock(businessKey, lockMaxTime, tryTimeout, timeUnit);
}
public boolean redisLock(String businessKey, long lockMaxTime, long tryTimeout, TimeUnit timeUnit) {
tryTimeout = System.currentTimeMillis() + timeUnit.toMillis(tryTimeout);
lockMaxTime = timeUnit.toMillis(lockMaxTime);
return redisLock(businessKey, lockMaxTime, tryTimeout);
}
/**
* 获取redis分布式锁 (默认超时时间)
*
* @param keyPrefix redis锁 key前缀
* @param key key
* @return 是否获取到锁资源
*/
@Deprecated
public boolean redisLock(String keyPrefix, String key) {
String businessKey = getLockKey(keyPrefix, key);
return redisLock(businessKey);
}
public boolean redisLock(String businessKey) {
long endTime = System.currentTimeMillis() + LOCK_WAIT_MAX_TIME;
long lockMaxTime = TimeUnit.SECONDS.toMillis(this.lockMaxSeconds);
return redisLock(businessKey, lockMaxTime, endTime);
}
/**
* 获取redis分布式锁 (默认超时时间)
* @param businessKey 业务key
* @param lockMaxTime 锁占用时长
* @param endTime 结束时间
* @return 是否获取到锁资源
*/
private boolean redisLock(String businessKey, long lockMaxTime, long endTime) {
String lockKey = generateLockKey(businessKey);
logger.debug("redisLock businessKey:{}, lockKey:{}, lockMaxTime:{}, endTime:{}", businessKey, lockKey, lockMaxTime, endTime);
//可重入锁判断
if (isReentrantLock(lockKey)) {
logger.debug("redisLock lockKey:{}, threadName:{}, isReentrantLock true", lockKey, Thread.currentThread().getName());
return Boolean.TRUE;
}
RedisCallback<Boolean> callback = (connection) -> connection.set(
lockKey.getBytes(StandardCharsets.UTF_8),
businessKey.getBytes(StandardCharsets.UTF_8),
Expiration.milliseconds(lockMaxTime),
RedisStringCommands.SetOption.SET_IF_ABSENT);
//在timeout时间内仍未获取到锁，则获取失败
while (System.currentTimeMillis() < endTime) {
if (stringRedisTemplate.execute(callback)) {
redisLockReentrant.set(lockKey);
logger.debug("redisLock getKey lockKey:{}, ", lockKey);
return true;
}
try {
Thread.sleep(100);
} catch (InterruptedException e) {
logger.error("获取redis分布式锁出错", e);
Thread.currentThread().interrupt();
}
}
logger.debug("redisLock meiyoukey lockKey:{}, ", lockKey);
return false;
}
/**
* 释放分布式锁
*
* @param keyPrefix redis锁 key前缀
* @param key key
*/
@Deprecated
public Boolean redisUnLock(String keyPrefix, String key) {
String lockKey = getLockKey(keyPrefix, key);
return redisUnLock(lockKey);
}
public Boolean redisUnLock(String businessKey) {
String lockKey = generateLockKey(businessKey);
RedisCallback<Boolean> callback = (connection) -> connection.eval(
RELEASE_LOCK_SCRIPT.getBytes(),
ReturnType.BOOLEAN, 1,
lockKey.getBytes(StandardCharsets.UTF_8),
businessKey.getBytes(StandardCharsets.UTF_8));
//清空 ThreadLocal
redisLockReentrant.remove();
Boolean execute = stringRedisTemplate.execute(callback);
logger.debug("redisUnLock execute lockKey:{}, ", lockKey);
return execute;
}
private String getLockKey(String keyPrefix, String key) {
return keyPrefix + "-" + key;
}
/**
* 是否为重入锁
*/
private boolean isReentrantLock(String lockKey) {
String originValue = redisLockReentrant.get();
String redisValue = stringRedisTemplate.opsForValue().get(lockKey);
return StringUtils.isNotBlank(originValue) && originValue.equals(redisValue);
}
/**
* 生成规则要求的 key
* xxx:redisLock:${appName}:${classSimpleName}:${methodName}:${businessKey}
* @param businessKey 业务key
* @return key
*/
private String generateLockKey(String businessKey) {
StackTraceElement[] stackTrace = Thread.currentThread().getStackTrace();
String classSimpleName = StringUtils.EMPTY;
String methodName = StringUtils.EMPTY;
for (StackTraceElement traceElement : stackTrace) {
String itemClassName = traceElement.getClassName();
//如果是当前类或者stack类 continue;
if (itemClassName.equals(this.getClass().getName()) || itemClassName.equals(StackTraceElement.class.getName())) {
continue;
}
char[] cs=itemClassName.substring(itemClassName.lastIndexOf(".")+1).toCharArray();
cs[0]+=32;
//一直找，找到被spring管理的类。
Object target;
try {
target = applicationContext.getBean(String.valueOf(cs));
} catch (NoSuchBeanDefinitionException e) {
continue;
}
//如果是代理类，找到实际类
if (AopUtils.isAopProxy(target) && target instanceof Advised) {
Advised advised = (Advised) target;
try {
target = advised.getTargetSource().getTarget();
} catch (Exception e) {
continue;
}
}
if (Objects.nonNull(target)) {
classSimpleName = target.getClass().getSimpleName();
methodName = traceElement.getMethodName();
break;
}
}
return REDIS_LOCK_KEY_PREFIX.concat(REDIS_NAMESPACE_PREFIX).concat(appName.toLowerCase())
.concat(REDIS_NAMESPACE_PREFIX).concat(classSimpleName)
.concat(REDIS_NAMESPACE_PREFIX).concat(methodName)
.concat(REDIS_NAMESPACE_PREFIX).concat(businessKey);
}
}

复制代码

com.iceicepip.project.common.redis.CustomRedisConfig

复制代码

com.iceicepip.project.common.redis.CustomRedisConfig

复制代码

@Autowired
@Qualifier("xxxRedis")
private CustomRedisService xxxRedisService;
@Autowired
@Qualifier("userRedis")
private CustomRedisService userRedisService;

复制代码

@Autowired
@Qualifier("userRedisTemplate")
private RedisTemplate userRedisTemplate;
@Autowired
@Qualifier("xxxStringRedisTemplate")
private StringRedisTemplate xxxStringRedisTemplate;
@Autowired
@Qualifier("xxxRedisTemplate")
private RedisTemplate xxxRedisTemplate;

复制代码