Redis实现库存，分布式锁，和集群

深碍√TFBOYSˉ_ 2023-07-07 05:47 33阅读 0赞

# Redis: #

### 一. redis实现库存增减： ###

使用lua脚本：

alloc.lua: (分配库存脚本)

--[[资源分配 分配资源--]]
 
 --[[参数--]]
 local TOTAL_FILED = "COUNT"
 local ALLOC_FILED = "ALLOC"
 
 local identifier = KEYS[1]
 local acquireCount = tonumber(ARGV[1])
 -- 总数量
 local totalCount = redis.call("HGET", identifier, TOTAL_FILED)
 -- 已申请数量
 local allocCount = redis.call("HGET", identifier, ALLOC_FILED)
 
 --[[资源没有初始化的时候申请,返回分配失败]]
 if false == totalCount then
 return false
 end
 --[[第一次申请初始化已经申请数量为0]]
 if false == allocCount then
 allocCount = 0
 redis.call("HSET", identifier, ALLOC_FILED, 0)
 end
 
 local totalCountNumber = tonumber(totalCount)
 allocCount = tonumber(allocCount)
 -- 将申请总数加当次申请数
 if allocCount + acquireCount <= totalCountNumber then
 redis.call("HINCRBY", identifier, ALLOC_FILED, acquireCount)
 return true
 else
 return false
 end

释放库存脚本：release.lua

--[[资源分配  释放资源--]]
    
    --[[分配资源数量字段--]]
    local ALLOC_FILED = "ALLOC"
    
    local identifier = KEYS[1]
    local releaseCount = tonumber(ARGV[1])
    
    local allocCount = tonumber(redis.call("HGET", identifier, ALLOC_FILED))
    
    if releaseCount > allocCount then
        redis.call("HSET", identifier, 0)
    else
        redis.call("HGET", identifier, allocCount - releaseCount)
    end
    
    return true

资源分配中心接口：

/** * 资源分配中心 * */
    public interface ResourcesAllocCenter { 
        /** * 申请资源 * * @param resourcesIdentifier 资源标识 * @param count 资源数量 * @return true 申请成功 false 申请失败 */
        boolean alloc(String resourcesIdentifier, Integer count);
    
        /** * 设置资源数量 * * @param resourcesIdentifier 资源标识 * @param count 数量 * @return 是否初始化成功 */
        boolean setCount(String resourcesIdentifier, Integer count);
    
        /** * 获取资源数量 * * @param resourcesIdentifier 资源标识 * @return 资源数量 */
        int getCount(String resourcesIdentifier);
    
        /** * 设置资源已分配数量 * * @param resourcesIdentifier 资源标识 * @param allocCount 已分配数量 * @return 是否设置成功 */
        boolean setAllocCount(String resourcesIdentifier, Integer allocCount);
    
        /** * 获取资源已分配数量 * * @param resourcesIdentifier 资源标识符 * @return 资源已分配数量 */
        int getAllocCount(String resourcesIdentifier);
    
    
        /** * 释放资源 * * @param resourcesIdentifier 资源标识 * @param count 释放资源数量 * @return 是否释放成功 */
        boolean release(String resourcesIdentifier, Integer count);
    
        /** * 删除资源 * * @param resourcesIdentifier 资源标识 * @return 是否删除成功 */
        boolean remove(String resourcesIdentifier);
    }

资源分配中心实现：

package com.test.spring.tools.resources;
 
 
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
 import org.springframework.core.io.ClassPathResource;
 import org.springframework.data.redis.core.RedisTemplate;
 import org.springframework.data.redis.core.script.RedisScript;
 import org.springframework.scripting.support.ResourceScriptSource;
 import org.springframework.stereotype.Component;
 
 import java.io.IOException;
 import java.util.Collections;
 import java.util.Objects;
 
 /** * redis 实现资源分配在中心 */
 
 @Slf4j
 @Component
 @ConditionalOnProperty(prefix = "spring.redis", value = "database")
 public class RedisResourcesAllocCenter implements ResourcesAllocCenter { 
 
 /** * 资源总数量存储的字段 */
 private static final String TOTAL_FILED = "COUNT";
 /** * 资源已经分配的数量存储的字段 */
 private static final String ALLOC_FILED = "ALLOC";
 
 /** * 获取资源脚本 */
 private static RedisScript<Boolean> ALLOC_SCRIPT;
 /** * 释放资源脚本 */
 private static RedisScript<Boolean> RELEASE_SCRIPT;
 
 //初始化加载脚本文件
 static { 
 try { 
 ALLOC_SCRIPT = RedisScript.of(new ResourceScriptSource(new ClassPathResource("redis/alloc.lua")).getScriptAsString(), Boolean.class);
 RELEASE_SCRIPT = RedisScript.of(new ResourceScriptSource(new ClassPathResource("redis/release.lua")).getScriptAsString(), Boolean.class);
 } catch (IOException e) { 
 log.error("加载配置文件初出错");
 }
 }
 
 private final RedisTemplate<String, Object> redisTemplate;
 
 public RedisResourcesAllocCenter(RedisTemplate<String, Object> redisTemplate) { 
 this.redisTemplate = redisTemplate;
 }
 
 @Override
 public boolean alloc(String resourcesIdentifier, Integer count) { 
 log.info("alloc resources, resourcesIdentifier: {}, count: {}", resourcesIdentifier, count);
 Boolean result = redisTemplate.execute(ALLOC_SCRIPT, Collections.singletonList(resourcesIdentifier), String.valueOf(count));
 log.info("alloc resources, resourcesIdentifier: {}, count: {}, result:{}", resourcesIdentifier, count, result);
 return result != null && result;
 }
 
 @Override
 public boolean setCount(String resourcesIdentifier, Integer count) { 
 log.info("set resources count. resourcesIdentifier: {}, count:{}.", resourcesIdentifier, count);
 redisTemplate.opsForHash().put(resourcesIdentifier, TOTAL_FILED, String.valueOf(count));
 return true;
 }
 
 @Override
 public int getCount(String resourcesIdentifier) { 
 log.info("get resources count. resourcesIdentifier: {}.", resourcesIdentifier);
 Object result = redisTemplate.opsForHash().get(resourcesIdentifier, TOTAL_FILED);
 if (Objects.isNull(result)) { 
 return 0;
 }
 log.info("set resources count. resourcesIdentifier: {}.", resourcesIdentifier);
 return Integer.valueOf(String.valueOf(result));
 }
 
 @Override
 public boolean setAllocCount(String resourcesIdentifier, Integer count) { 
 log.info("set resources alloc count. resourcesIdentifier: {}, count:{}.", resourcesIdentifier, count);
 redisTemplate.opsForHash().put(resourcesIdentifier, ALLOC_FILED, String.valueOf(count));
 return true;
 }
 
 @Override
 public int getAllocCount(String resourcesIdentifier) { 
 log.info("get resources alloc count. resourcesIdentifier: {}.", resourcesIdentifier);
 Object result = redisTemplate.opsForHash().get(resourcesIdentifier, ALLOC_FILED);
 if (Objects.isNull(result)) { 
 return 0;
 }
 log.info("set resources alloc count. resourcesIdentifier: {}.", resourcesIdentifier);
 return Integer.valueOf(String.valueOf(result));
 }
 
 @Override
 public boolean release(String resourcesIdentifier, Integer count) { 
 log.info("release resources, resourcesIdentifier: {}, count: {}", resourcesIdentifier, count);
 Boolean result = redisTemplate.execute(RELEASE_SCRIPT, Collections.singletonList(resourcesIdentifier), String.valueOf(count));
 log.info("release resources, resourcesIdentifier: {}, count: {}, result:{}", resourcesIdentifier, count, result);
 return result != null && result;
 }
 
 @Override
 public boolean remove(String resourcesIdentifier) { 
 log.info("remove resources. resourcesIdentifier: {}.", resourcesIdentifier);
 try { 
 redisTemplate.delete(resourcesIdentifier);
 } catch (Exception e) { 
 log.info("remove resources failure. resourcesIdentifier: {}.", resourcesIdentifier);
 return false;
 }
 log.info("remove resources success. resourcesIdentifier: {}.", resourcesIdentifier);
 return true;
 }
 }

初始化库存：

实现CommandLineRunner接口，会在springboot启动时执行，如果有多个，可以使用order注解指定执行顺序

package com.test.medusa.manage.runner;
 
 import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
 import com.egls.cashloan.medusa.core.enums.common.SysStateEnum;
 import com.egls.cashloan.medusa.core.primitive.prize.entity.PrizeConfig;
 import com.egls.cashloan.medusa.core.primitive.prize.service.PrizeConfigService;
 import com.egls.cashloan.medusa.core.primitive.prize.service.PrizeDrawOrderService;
 import com.egls.cashloan.medusa.core.stocks.RedisPrizeStocksCenter;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.boot.CommandLineRunner;
 import org.springframework.stereotype.Component;
 
 import javax.annotation.Resource;
 import java.util.List;
 
 
 @Slf4j
 @Component
 public class PrizeRunner implements CommandLineRunner { 
 
 @Resource
 private PrizeConfigService prizeConfigService;
 
 @Resource
 private RedisPrizeStocksCenter redisPrizeStocksCenter;
 
 @Resource
 private PrizeDrawOrderService prizeDrawOrderService;
 
 @Override
 public void run(String... args) throws Exception { 
 LambdaQueryWrapper<PrizeConfig> lambdaQueryWrapper = new LambdaQueryWrapper<>();
 lambdaQueryWrapper.eq(PrizeConfig::getState, SysStateEnum.ACTIVE.getValue());
 List<PrizeConfig> prizeConfigs = prizeConfigService.list(lambdaQueryWrapper);
 prizeConfigs.forEach(prizeConfig -> { 
 long total = prizeConfig.getInitialAmount() + prizeConfig.getAppendAmount();
 redisPrizeStocksCenter.setCount(prizeConfig.getCode(), (int) total);
 });
 
 }
 }

申请库存代码；

@Override
 @Transactional(rollbackFor = Exception.class)
 public Map<String, Object> prizeDraw(UserBaseInfo userBaseInfo) { 
 Map<String, Object> result = new HashMap<>();
 result.put("code", "20");
 result.put("msg", "很遗憾未中奖");
 
 // 2. 查询出该用户级别对应的所有奖品及概率,并按照概率排序
 LambdaQueryWrapper<PrizeConfig> prizeConfigLambdaQueryWrapper = new LambdaQueryWrapper<>();
 prizeConfigLambdaQueryWrapper.eq(PrizeConfig::getState, SysStateEnum.ACTIVE.getValue());
 List<PrizeConfig> prizeConfigs = this.list(prizeConfigLambdaQueryWrapper);
 
 List<PrizeConfigTemp> prizeConfigTemps = new ArrayList<>();
 int total = 1000000;
 AtomicInteger sum = new AtomicInteger();
 prizeConfigs.forEach(prizeConfig -> { 
 int rate = new BigDecimal(prizeConfig.getRate()).multiply(new BigDecimal("10000")).intValue();
 sum.addAndGet(rate);
 PrizeConfigTemp prizeConfigTemp = new PrizeConfigTemp(prizeConfig.getCode(), prizeConfig.getName(),prizeConfig.getType(),prizeConfig.getMoney(), rate);
 prizeConfigTemps.add(prizeConfigTemp);
 });
 if(total > sum.get()){ 
 PrizeConfigTemp prizeConfigTemp = new PrizeConfigTemp("null", "null","null",BigDecimal.ZERO, total- sum.get());
 prizeConfigTemps.add(prizeConfigTemp);
 }
 PrizeConfigTemp prizeConfigTemp = MathUtils.turnTableBet(prizeConfigTemps);
 boolean allocOne = true;
 if (prizeConfigTemp == null) { 
 log.info("中奖出现空，请检测配置");
 } else { 
 if(StringUtils.equals("null",prizeConfigTemp.getCode())){ 
 log.info("未中奖" + prizeConfigTemp.toString());
 }else { 
 log.info("中奖结果" + prizeConfigTemp.toString());
 // 此处为申请库存代码
 allocOne = redisPrizeStocksCenter.allocOne(prizeConfigTemp.getCode());
 if (allocOne) { 
 PrizeConfig prizeConfig = this.getByCode(prizeConfigTemp.getCode());
 result.put("code", "10");
 result.put("msg", prizeConfig.getName());
 result.put("data", prizeConfig);
 } else { 
 log.info("库存不足" + prizeConfigTemp.toString());
 }
 }
 }
 
 //1.扣除积分，目前固定 1 分
 StringBuilder remark = new StringBuilder();
 remark.append("抽奖消耗积分，奖品:");
 remark.append(prizeConfigTemp != null && allocOne && !StringUtils.equals("null",prizeConfigTemp.getCode())? prizeConfigTemp.getName() : "未中奖");
 // 积分调整,积分传负
 userPointsApi.pointsAdjust(userBaseInfo.getUserId(), UserPointsOperationTypeEnum.WIN_PRIZE, new BigDecimal("-1"),null, LocalDateTime.now(),remark.toString());
 /** * 消耗积分数 */
 result.put("pointsConsumed", 1);
 
 // 插入中奖订单表
 PrizeDrawOrder order = new PrizeDrawOrder();
 int invalidDuration = 3;
 if(systemConfigCache.getInt("invalid_duration").isPresent()){ 
 invalidDuration = systemConfigCache.getInt("invalid_duration").get();
 }
 order.setCreateTime(LocalDateTime.now());
 order.setGradeCode(userBaseInfo.getGradeCode());
 order.setGradeName(userBaseInfo.getGradeName());
 order.setMobile(userBaseInfo.getMobile());
 order.setName(userBaseInfo.getName());
 order.setNickname(userBaseInfo.getNickname());
 order.setUserId(userBaseInfo.getUserId());
 if (prizeConfigTemp != null && allocOne && !StringUtils.equals("null",prizeConfigTemp.getCode())) { 
 order.setOrderNo(PayOrderNoUtil.getOcpOrderNo());
 order.setPrizeCode(prizeConfigTemp.getCode());
 order.setPrizeName(prizeConfigTemp.getName());
 order.setPrizeType(prizeConfigTemp.getType());
 order.setPrizeMoney(prizeConfigTemp.getMoney());
 order.setState("10");
 LocalDateTime invalidDate = LocalDateTime.of(LocalDateTime.now().plusDays(invalidDuration - 1).toLocalDate(), LocalTime.of(23,59,59));
 order.setInvalidDate(invalidDate);
 
 // 发送抽中奖品短信：
 LambdaQueryWrapper<SmsTpl> smsTplQueryWrapper = new LambdaQueryWrapper<>();
 smsTplQueryWrapper.eq(SmsTpl::getType, "WIN_PRIZE");
 SmsTpl smsTpl = smsTplService.getOne(smsTplQueryWrapper);
 if(smsTpl != null){ 
 String content = String.format(smsTpl.getTpl(), prizeConfigTemp.getName());
 log.info("开始发送用户{}，中奖通知短信，内容为：{}",userBaseInfo.getUserId(),content);
 smsLogService.sendSmsRightNow(userBaseInfo.getUserId(),userBaseInfo.getMobile(),smsTpl.getType(),content,order.getOrderNo());
 }
 } else { 
 order.setState("00");
 }
 prizeDrawOrderService.save(order);
 return result;
 }

### 二. redis 实现分布式锁： ###

依然使用lua脚本实现：

加锁脚本：lock.lua

--[[分布式锁,获取锁实现,设置值并设置过期时间--]]
    
    --[[参数--]]
    local key_name = KEYS[1]
    local expire_time = tonumber(ARGV[1])
    local lock_id_key_name = ARGV[2]
    
    --[[初始化锁ID--]]
    local exists = redis.call("EXISTS", lock_id_key_name)
    if 0 == exists then
        redis.call("SET", lock_id_key_name, 10000)
    end
    
    --[[获取锁，获取成功返回锁ID，获取失败返回空字符串--]]
    local lock_id = redis.call("INCR", lock_id_key_name)
    local set_result = redis.call("SETNX", key_name, lock_id)
    if 1 == set_result then
        redis.call("EXPIRE", key_name, expire_time)
        return lock_id .. ''
    else
        return ''
    end

释放锁脚本：unlock.lua

--[[分布式锁,释放锁实现，如果key不存在则释放成功--]]
    
    --[[参数--]]
    local key_name = KEYS[1]
    local key_value = ARGV[1]
    
    --[[如果key已经不存在则,释放成功--]]
    local exists = redis.call("EXISTS", key_name)
    if 0 == exists then
        return 1
    end
    
    --[[如果key存在,比较当前值和传递过来的keyId是否一致
        一致删除key.
        不一致则不做任何操作,此时所可能已经被其它任务获取,也释放成功
    --]]
    local get_result = redis.call("GET", key_name)
    if key_value == get_result then
        redis.call("DEL", key_name)
    end
    
    return 1

上锁解锁，都要设置唯一的标识：

分布式锁代码：

package com.test.spring.tools.lock;
 
 import lombok.extern.slf4j.Slf4j;
 import org.apache.commons.lang3.StringUtils;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
 import org.springframework.core.io.ClassPathResource;
 import org.springframework.data.redis.core.RedisTemplate;
 import org.springframework.data.redis.core.script.RedisScript;
 import org.springframework.scripting.support.ResourceScriptSource;
 import org.springframework.stereotype.Component;
 
 import java.io.IOException;
 import java.time.LocalDateTime;
 import java.time.temporal.ChronoUnit;
 import java.util.Collections;
 import java.util.Optional;
 import java.util.concurrent.TimeUnit;
 
 /** * * Redis 分布式锁,目前提供几个基础方法: * 获取锁: 调用获取锁方法之后，需要对返回结果进行判断，如果不为空则表示获取成功，获取锁成功之后会返回锁拥有者的Id * {@link #lock(String)} 使用默认过期时间 100 秒的lock() * {@link #lock(String, long)} 需要指定过期时间的 lock() * {@link #lockBlocked(String, long)} 阻塞获取直到获取成功 * {@link #lockBlockedWithTimeOut(String, long)} 在指定的时间内尝试获取,如果到达指定时间没有获取成功会返回获取失败。 * {@link #lockBlockedWithTimeOut(String, long, long)} 阻塞指定时间，同事指定锁过期时间 * * 释放锁: * {@link #unLock(String, String)} :释放锁操作不需要返回值，每次调用都返回成功，存在以下几种情况: * * 业务A释放锁的时候锁已经因为过期而释放 * * 业务A释放的时候锁已经已为过期而释放，并且此时又被业务B获取，业务A的锁已经释放 * * 业务A的锁没有因为过期而被释放，此时删除对应的 key。 * * 注意：对于前两种情况多是因为锁的过期设置的太短而导致的,需要业务获取锁的时候指定合适的过期时间。 * * * */
 @Slf4j
 @Component
 @ConditionalOnProperty(prefix = "spring.redis", value = "database")
 public class RedisDistributedLock { 
 
 private static final long WAIT_TIME_MILL_SECONDS = 100;
 
 private final RedisTemplate<String, Object> redisTemplate;
 
 public RedisDistributedLock(RedisTemplate<String, Object> redisTemplate) { 
 this.redisTemplate = redisTemplate;
 }
 
 private static final String LOCK_PREFIX = "{LOCK}_RS_";
 private static final String LOCK_ID = "{LOCK}_ID";
 
 /** * 默认key过期时间，单位秒 */
 private static final Long DEFAULT_KEY_EXPIRE_TIME = 100L;
 
 /** * /** * Redis脚本 */
 private static RedisScript<String> LOCK_SCRIPT;
 private static RedisScript<Boolean> UN_LOCK_SCRIPT;
 
 //初始化加载脚本文件
 static { 
 try { 
 LOCK_SCRIPT = RedisScript.of(new ResourceScriptSource(new ClassPathResource("redis/lock.lua")).getScriptAsString(), String.class);
 UN_LOCK_SCRIPT = RedisScript.of(new ResourceScriptSource(new ClassPathResource("redis/unlock.lua")).getScriptAsString(), Boolean.class);
 } catch (IOException e) { 
 log.error("加载配置文件初出错");
 }
 }
 
 /** * 获取锁，只获取一次 * 使用默认过期时间 {@link #DEFAULT_KEY_EXPIRE_TIME} * * @param resourceIdentify 资源标识 * @return 锁Id，用于释放锁,返回为空表示所获取失败 */
 public Optional<String> lock(String resourceIdentify) { 
 log.info("获取资源锁，使用默认过期时间，资源标识:{}, 过期时间:{}", resourceIdentify, DEFAULT_KEY_EXPIRE_TIME);
 return lock(resourceIdentify, DEFAULT_KEY_EXPIRE_TIME);
 }
 
 /** * 阻塞获取锁，直至获取成功 * * @param resourceIdentify 资源标识 * @param expireTime 锁自动释放时间 * @return 锁Id，用于释放锁,返回为空表示所获取失败 */
 public Optional<String> lockBlocked(String resourceIdentify, long expireTime) { 
 log.info("阻塞获取资源锁，资源标识:{}, 过期时间:{}", resourceIdentify, expireTime);
 Optional<String> optional;
 try { 
 int times = 0;
 while (true) { 
 times++;
 optional = lock(resourceIdentify, DEFAULT_KEY_EXPIRE_TIME);
 if (optional.isPresent()) { 
 log.info("阻塞获取资源锁,获取成功.资源标识:{}, 过期时间:{},尝试次数:{}", resourceIdentify, expireTime, times);
 return optional;
 }
 log.info("阻塞获取资源锁,获取成功.资源标识:{}, 过期时间:{},尝试次数:{}", resourceIdentify, expireTime, times);
 TimeUnit.MILLISECONDS.sleep(WAIT_TIME_MILL_SECONDS);
 }
 } catch (Exception e) { 
 log.error("阻塞获取资源锁失败，资源标识:{}, 过期时间:{}", resourceIdentify, expireTime);
 log.error(e.getMessage(), e);
 }
 return Optional.empty();
 }
 
 /** * 阻塞获取锁,默认锁持有时间100秒 * * @param resourceIdentify 资源标识符 * @param timeOut 获取锁超时时间,单位毫秒 * @return 获取不成功返回空, 获取成功返回锁标识 */
 public Optional<String> lockBlockedWithTimeOut(String resourceIdentify, long timeOut) { 
 return lockBlockedWithTimeOut(resourceIdentify, DEFAULT_KEY_EXPIRE_TIME, timeOut);
 }
 
 /** * 阻塞获取锁 * * @param resourceIdentify 资源标识符 * @param expireTime 锁过期时间，单位秒 * @param timeOut 获取锁超时时间,单位毫秒 * @return 获取不成功返回空, 获取成功返回锁标识 */
 public Optional<String> lockBlockedWithTimeOut(String resourceIdentify, long expireTime, long timeOut) { 
 log.info("阻塞获取资源锁，过期时间，资源标识:{}, 过期时间:{}，超时时间:{}", resourceIdentify, expireTime, timeOut);
 Optional<String> optional;
 try { 
 int times = 0;
 LocalDateTime endTime = LocalDateTime.now().plus(timeOut, ChronoUnit.MILLIS);
 while (LocalDateTime.now().isBefore(endTime)) { 
 times++;
 optional = lock(resourceIdentify, expireTime);
 if (optional.isPresent()) { 
 log.info("阻塞获取资源锁,获取成功.资源标识:{}, 过期时间:{},尝试次数:{}，超时时间:{}", resourceIdentify, expireTime, times, timeOut);
 return optional;
 }
 log.info("阻塞获取资源锁,获取失败等待重试.资源标识:{}, 过期时间:{},尝试次数:{}，超时时间:{}", resourceIdentify, expireTime, times, timeOut);
 TimeUnit.MILLISECONDS.sleep(WAIT_TIME_MILL_SECONDS);
 }
 } catch (Exception e) { 
 log.error("阻塞获取资源锁失败，过期时间，资源标识:{}, 过期时间:{}", resourceIdentify, expireTime);
 log.error(e.getMessage(), e);
 }
 return Optional.empty();
 }
 
 
 /** * 获取锁 * * @param resourceIdentify 资源标识 * @param expireTime 过期时间, 单位:秒 * @return 锁Id，用于释放锁,返回为空表示所获取失败 */
 public Optional<String> lock(String resourceIdentify, long expireTime) { 
 log.info("获取资源锁，使用自定义过期时间，资源标识:{}, 过期时间:{}", resourceIdentify, expireTime);
 String result = redisTemplate.execute(LOCK_SCRIPT, Collections.singletonList(getKey(resourceIdentify)), String.valueOf(expireTime), getLockIdKeyName());
 if (StringUtils.isEmpty(result)) { 
 log.info("获取资源锁，获取失败。使用自定义过期时间，资源标识:{}, 锁ID:{}, 过期时间:{}", resourceIdentify, result, expireTime);
 return Optional.empty();
 } else { 
 log.info("获取资源锁，获取成功。使用自定义过期时间，资源标识:{}, 锁ID:{}, 过期时间:{}", resourceIdentify, result, expireTime);
 return Optional.of(result);
 }
 }
 
 
 /** * 释放锁 * * @param resourceIdentify 资源标识ID * @param lockId 获取锁成功之后返回的锁ID */
 public void unLock(String resourceIdentify, String lockId) { 
 log.info("释放锁，资源标识：{}", resourceIdentify);
 redisTemplate.execute(UN_LOCK_SCRIPT, Collections.singletonList(getKey(resourceIdentify)), lockId);
 }
 
 
 /** * 删除资源锁，慎重使用 * * @param resourceIdentify 资源标识符 */
 @Deprecated
 public void delete(String resourceIdentify) { 
 log.info("删除锁，资源标识：{}", resourceIdentify);
 redisTemplate.delete(getKey(resourceIdentify));
 }
 
 /** * 获取资源ID，使用{}确保分布式锁的key，被散列到集群的同一台机器上 * * @param resourceIdentify 资源表示 * @return key */
 private String getKey(String resourceIdentify) { 
 return String.format("%s%s", LOCK_PREFIX, resourceIdentify);
 }
 
 /** * 获取锁ID的key name，用于redis记录 lock id * * @return lock id key name */
 private String getLockIdKeyName() { 
 return LOCK_ID;
 }
 }

redis配置：

package com.test.spring.tools.config.redis;
 
 import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
 import org.springframework.data.redis.connection.RedisConnectionFactory;
 import org.springframework.data.redis.core.*;
 import org.springframework.data.redis.serializer.StringRedisSerializer;
 
 /** * Redis 配置 * */
 @Configuration
 @ConditionalOnProperty(prefix = "spring.redis", value = "database")
 public class RedisTemplateConfig { 
 
 private final RedisConnectionFactory factory;
 
 public RedisTemplateConfig(RedisConnectionFactory factory) { 
 this.factory = factory;
 }
 
 @Bean
 public RedisTemplate<String, Object> redisTemplate() { 
 RedisTemplate<String, Object> redisTemplate = new RedisTemplate<>();
 redisTemplate.setKeySerializer(new StringRedisSerializer());
 redisTemplate.setValueSerializer(new StringRedisSerializer());
 redisTemplate.setHashKeySerializer(new StringRedisSerializer());
 redisTemplate.setHashValueSerializer(new StringRedisSerializer());
 redisTemplate.setConnectionFactory(factory);
 return redisTemplate;
 }
 
 @Bean
 public HashOperations<String, String, Object> hashOperations(RedisTemplate<String, Object> redisTemplate) { 
 return redisTemplate.opsForHash();
 }
 
 @Bean
 public ValueOperations<String, String> valueOperations(RedisTemplate<String, String> redisTemplate) { 
 return redisTemplate.opsForValue();
 }
 
 @Bean
 public ListOperations<String, Object> listOperations(RedisTemplate<String, Object> redisTemplate) { 
 return redisTemplate.opsForList();
 }
 
 @Bean
 public SetOperations<String, Object> setOperations(RedisTemplate<String, Object> redisTemplate) { 
 return redisTemplate.opsForSet();
 }
 
 @Bean
 public ZSetOperations<String, Object> zSetOperations(RedisTemplate<String, Object> redisTemplate) { 
 return redisTemplate.opsForZSet();
 }
 }

package com.test.spring.tools.bean;
 
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.beans.BeansException;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnWebApplication;
 import org.springframework.context.ApplicationContext;
 import org.springframework.context.ApplicationContextAware;
 import org.springframework.core.env.Profiles;
 import org.springframework.stereotype.Component;
 import org.springframework.util.StreamUtils;
 
 import java.io.File;
 import java.io.IOException;
 import java.io.InputStream;
 import java.util.Objects;
 
 /** * 应用程序工具类 * */
 @Slf4j
 @Component
 @ConditionalOnWebApplication
 public class WebApplicationUtil implements ApplicationContextAware { 
 
 private static final String DEV_PROFILE = "dev";
 
 private ApplicationContext applicationContext;
 
 private String contentPath;
 
 /** * 根据类名获取Bean对象 * * @param tClass 类名 * @param <T> 类型 * @return 对应的Bean实例 */
 public <T> T getBean(Class<T> tClass) { 
 return applicationContext.getBean(tClass);
 }
 
 public Object getBean(String beanName) { 
 return applicationContext.getBean(beanName);
 }
 
 /** * 获取资源文件位置 * * @return 资源文件目录 */
 public String getResourcePath() throws IOException { 
 return this.contentPath;
 }
 
 
 /** * 读取class path 下的文件 * * @param filePath 文件路径 * @return 文件数据 */
 public byte[] readClassPathFile(String filePath) throws IOException { 
 InputStream stream = Thread.currentThread().getContextClassLoader().getResourceAsStream(filePath);
 if (Objects.isNull(stream)) { 
 String message = String.format("读取 class path 下文件，资源文件不存在，文件: %s", filePath);
 log.info(message);
 throw new RuntimeException(message);
 }
 return StreamUtils.copyToByteArray(stream);
 }
 
 /** * 根据相对路径获取容器运行路径的绝对路径 * * @param relativePath 相对路径 * @return 容器运行路径的绝对路径 */
 public String getAbsoluteContentPath(String relativePath) { 
 return this.contentPath + File.separator + relativePath;
 }
 
 @Override
 public void setApplicationContext(ApplicationContext applicationContext) throws BeansException { 
 this.applicationContext = applicationContext;
 try { 
 this.contentPath = applicationContext.getResource("").getFile().getAbsolutePath();
 } catch (Exception e) { 
 log.error(String.format("WebApplicationUtil 初始化异常, 异常信息: %s", e.getMessage()), e);
 throw new RuntimeException("系统启动初始化路径异常", e);
 }
 log.info("WebApplicationUtil content path: {}", contentPath);
 }
 
 public boolean devIsActive() { 
 return applicationContext.getEnvironment().acceptsProfiles(Profiles.of(DEV_PROFILE));
 }
 }

### 三. redis 安装 ###

redis的安装非常简单，打开redis官网http://redis.io

下载一个最新版本的安装包，如redis-version.tar.gz

解压：tar zxvf redis-version.tar.gz

执行make命令（如果报错，比如缺少gcc等，一个一个解决即可）

如果是mac电脑，直接brew install redis

安装好之后，进行简单的配置， 打开redis.conf文件

port 6379		#指定端口号
    daemonize yes  	#指定后台运行

#### 3.1 redis 单点： ####

安装好之后，我们运行一下，启动命令为：

redishome/bin/redis-server path/to/redis.conf

假设我们没有配置后台运行，会看到一个启动日志，一个立方块。然后使用redis-cli 客户端，就可以使用redis命令操作了

当然我们可以使用代码来操作，常用的java客户端有：】

1.  jedis

<dependency>
 <groupId>redis.clients</groupId>
 <artifactId>jedis</artifactId>
 <version>3.0.0</version>
 <type>jar</type>
 <scope>compile</scope>
 </dependency>

Jedis jedis = new Jedis("localhost",6379);
    
    jedis.set("hello", "world");
    
    String value = jedis.get("hello");
    
    System.out.println(value); // get world
    
    jedis.del("hello");
    
    System.out.println(jedis.get("hello"));// get null

1.  spring-redis

public class Example { 
 
 // inject the actual template
 @Autowired
 private RedisTemplate<String, String> template;
 
 // inject the template as ListOperations
 // can also inject as Value, Set, ZSet, and HashOperations
 @Resource(name="redisTemplate")
 private ListOperations<String, String> listOps;
 
 public void addLink(String userId, URL url) { 
 listOps.leftPush(userId, url.toExternalForm());
 // or use template directly
 redisTemplate.boundListOps(userId).leftPush(url.toExternalForm());
 }
 }

1.  Lettuce:

<dependency>
 <groupId>io.lettuce</groupId>
 <artifactId>lettuce-core</artifactId>
 <version>5.1.5.RELEASE</version>
 </dependency>

public class LettuceTest { 
 
 public static void main(String[] args) { 
 RedisURI uri = new RedisURI();
 uri.setHost("myredishost");
 uri.setPort(6379);
 uri.setDatabase(0);
 RedisClient redisClient = RedisClient.create(uri);
 StatefulRedisConnection<String, String> connection = redisClient.connect();
 RedisCommands<String, String> syncCommands = connection.sync();
 
 syncCommands.set("testKey", "Hello, Redis!");
 System.out.println(syncCommands.get("testKey"));
 
 connection.close();
 redisClient.shutdown();
 }
 }

#### 3.2 redis 主从： ####

redis可以配置主从，主挂了之后，可以让从库接管，那么服务可以继续运行，有备无患。配置过程如下：

复制两个redis复制文件：

主配置文件不变，从配置文件加入：

## slave.conf
    port 6380
    slaveof 127.0.0.1 6379

分别启动两个redis

redis-server path/to/redis-master-slave/master.conf
    redis-server path/to/redis-master-slave/slave.conf

然后从maserset一个值，可以从从库查到，从从库set会报错，因为从库是只读的。

#### 3.3 哨兵sentinel: ####

主从模式虽然可以在主挂的时候，启动备份，但是必须我们人工介入，切换到从才可以。这时候哨兵模式就派上用场了。哨兵就是用来监控主从节点的健康情况。客户端连接上redis主从的时候，先连接sentinel,sentinel会告诉客户端主redis的地址是多少，然后客户端连接上redis并继续后续的操作，当主节点挂掉的时候，客户端就得不到连接而报错，客户端重新向sentinel询问master得之后，然后客户端得到了（新选举出来的master)

那么sentinel如何配置呢，我们先配置三个主从

mkdir redis-sentinel
    cd redis-sentinel
    cp redis/path/conf/redis.conf path/to/redis-sentinel/redis01.conf
    cp redis/path/conf/redis.conf path/to/redis-sentinel/redis02.conf
    cp redis/path/conf/redis.conf path/to/redis-sentinel/redis03.conf
    touch sentinel.conf

上面我们创建了三个redis文件，一个哨兵文件，我们将redis01设置为master，另外两个设置为slave

vim redis01.conf
    port 63791
    
    vim redis02.conf
    port 63792
    slaveof 127.0.0.1 63791
    
    vim redis03.conf
    port 63793
    slaveof 127.0.0.1 63791
    
    vim sentinel.conf
    daemonize yes
    port 26379
    sentinel monitor mymaster 127.0.0.1 63793 1   # 下面解释含义

解释一下哨兵配置：

mymaster        为主节点名字，可以随便取，后面程序里边连接的时候要用到
    127.0.0.1 63793 为主节点的 ip,port
    1               后面的数字 1 表示选举主节点的时候，投票数。1表示有一个sentinel同意即可升级为master

启动哨兵：

redis-server path/to/redis-sentinel/redis01.conf
    redis-server path/to/redis-sentinel/redis02.conf
    redis-server path/to/redis-sentinel/redis03.conf
    
    redis-server path/to/redis-sentinel/sentinel.conf --sentinel

模拟：

循环set值，把主节点shutdown:

从结果看出

开始正常操作redis，并设置了两次。  
主redis挂了，jedis得不到连接，报错了JedisConnectionException：Could not get a resource from the pool  
主redis没选好之前，程序持续报错。  
主redis选好了，程序正常运行，最后结束。  
我们看到最后一次运行设置的值是aajhbjagag,我们可以连接剩下的2台redis中的任意一台，get hello,结果肯定是一致的。

#### 3.4 redis cluster: ####

上面几种方式只是保证了数据备份一级高可用，目前为止我们的程序一直都是向1台redis写数据，其他的redis只是备份而已。实际场景中，单个redis节点可能不满足要求，因为：

单个redis并发优先；单个redis接受所有的数据，最终会导致内存太大，内存太大会导致rdb文件过大，从很大的rdb文件中同步恢复的数据会很慢。

所以我们需要redis cluster 即redis 集群。

redis集群是一个提供在多个redis节点间共享数据的程序集

redis集群并不支持处理多个keys的命令，因为这需要在不同的节点间移动数据，从而达不到想Redis那样的性能，在高负载情况下可能会导致不可预料的错误

redis集群通过分区来提供一定程度的可用性，在实际环境中当某个节点宕机或者不可达的情况下继续处理命令。redis集群优势：

自动分割数据到不同的节点上

整个集群的部分节点失败或者不可达的情况下能够继续处理命令。

为了配置一个redis cluster，我们需要哦至少准备六台，因为最小的集群，至少需要三个主节点。

主：redis01  从 redis02    slaveof redis01
    主：redis03  从 redis04    slaveof redis03
    主：redis05  从 redis06    slaveof redis05

mkdir redis-cluster
    cd redis-cluster
    mkdir redis01 到 redis06 6个文件夹
    cp redis.conf 到 redis01 ... redis06
    修改端口
    分别配置3组主从关系

启动集群：

redis-5.0.3/src/redis-cli --cluster create 127.0.0.1:6371 127.0.0.1:6372 127.0.0.1:6373 127.0.0.1:6374 127.0.0.1:6375 127.0.0.1:6376 --cluster-replicas 1

注意，这里使用的是ip:port，而不是 domain:port ，因为我在使用 localhost:6371 之类的写法执行的时候碰到错误：

执行成功之后，连接一台redis，执行 cluster info 会看到类似如下信息：

cluster_state:ok
    cluster_slots_assigned:16384
    cluster_slots_ok:16384
    cluster_slots_pfail:0
    cluster_slots_fail:0
    cluster_known_nodes:6
    cluster_size:3
    cluster_current_epoch:6
    cluster_my_epoch:1
    cluster_stats_messages_ping_sent:1515
    cluster_stats_messages_pong_sent:1506
    cluster_stats_messages_sent:3021
    cluster_stats_messages_ping_received:1501
    cluster_stats_messages_pong_received:1515
    cluster_stats_messages_meet_received:5
    cluster_stats_messages_received:3021

redis安装部分内容从知乎文章中整理，如果侵权请联系删除，库存和分布式锁部分结合自己生产实践。希望对大家有帮助