Photo @Christopher Campbell 

文 | 孔凡勇

概述

互联网应用发展到今天，从单体应用架构到 SOA 以及今天的微服务，随着微服务化的不断升级进化，服务和服务之间的稳定性变得越来越重要，分布式系统之所以复杂，主要原因是分布式系统需要考虑到网络的延时和不可靠，微服务很重要的一个特质就是需要保证服务幂等，保证幂等性很重要的前提需要分布式锁控制并发，同时缓存、降级和限流是保护微服务系统运行稳定性的三大利器。

redis.extend.hostName=127.0.0.1
redis.extend.port=6379
redis.extend.password=pwdcode
redis.extend.timeout=10000

redis.idempotent.enabled=true

接下来的篇幅，重点会介绍一下缓存、限流、分布式锁、幂等的使用方式。

缓存

缓存的使用可以说无处不在，从应用请求的访问路径来看，用户 user -> 浏览器缓存 -> 反向代理缓存-> WEB服务器缓存 -> 应用程序缓存 -> 数据库缓存等，几乎每条链路都充斥着缓存的使用，缓存最直白的解释就是“用空间换时间”的算法。缓存就是把一些数据暂时存放于某些地方，可能是内存，也有可能硬盘。总之，目的就是为了避免某些耗时的操作。我们常见的耗时的操作，比如数据库的查询、一些数据的计算结果，或者是为了减轻服务器的压力。其实减轻压力也是因查询或计算，虽然短耗时，但操作很频繁，累加起来也很长，造成严重排队等情况，服务器抗不住。

    public String get(String key);

    /**
     * 获取指定的key对应的对象,异常也会返回null
     *
     * @param key
     * @param clazz
     * @return
     */
    public <T> T get(String key, Class<T> clz);

    /**
     * 存储缓存数据,忽略过期时间
     *
     * @param key
     * @param value
     * @return
     */
    public <T extends Serializable> booleanput(String key, T value);

    /**
     * 存储缓存数据
     *
     * @param key
     * @param value
     * @param expiredTime
     * @param unit
     * @return
     */
    public <T extends Serializable> booleanput(String key, T value, int expiredTime, TimeUnit unit);

    /**
     * 基于key删除缓存数据
     *
     * @param key
     * @return
     */
    publicbooleaninvalid(String key);

get 方法针对 key 进行查询， put 存储缓存数据， invalid 删除缓存数据。

限流

 /**
     * 指定过期时间自增计数器，默认每次+1，非滑动窗口
     *
     * @param key 计数器自增key
     * @param expireTime 过期时间
     * @param unit  时间单位
     * @return
     */
    publiclongincrCount(String key, int expireTime, TimeUnit unit);

    /**
     * 指定过期时间自增计数器,单位时间内超过最大值rateThreshold返回true，否则返回false
     *
     * @param key 限流key
     * @param rateThreshold 限流阈值
     * @param expireTime 固定窗口时间
     * @param unit 时间单位
     * @return
     */
    publicbooleanrateLimit(final String key, finalint rateThreshold, int expireTime, TimeUnit unit);

基于 CacheEngine 的 rateLimit 方法可以实现限流， expireTime 只能设定固定窗口时间，非滑动窗口时间。

 /**
     * @param limitKey 限流KEY
     * @param resultSupplier 回调方法
     * @param rateThreshold 限流阈值
     * @param limitTime 限制时间段
     * @param blockDuration 阻塞时间段
     * @param unit 时间单位
     * @param errCodeEnum 指定限流错误码
     * @return
     */
    public <T> T execute(String limitKey, Supplier<T> resultSupplier, long rateThreshold, long limitTime,
                         long blockDuration, TimeUnit unit, ErrCodeEnum errCodeEnum){
        boolean blocked = tryAcquire(limitKey, rateThreshold, limitTime, blockDuration, unit);
        if (errCodeEnum != null) {
            AssertUtils.assertTrue(blocked, errCodeEnum);
        } else {
            AssertUtils.assertTrue(blocked, ExceptionEnumType.ACQUIRE_LOCK_FAIL);
        }

        return resultSupplier.get();
    }

另外 distributed-tools 组件还提供了注解 @RateLimit 的使用方式，具体注解 RateLimit 定义如下：

@Retention(RetentionPolicy.RUNTIME)
@Target(ElementType.METHOD)
@Documented
public @interface RateLimit {

    /**
     * 限流KEY
     */
    String limitKey();

    /**
     * 允许访问的次数，默认值MAX_VALUE
     */
    longlimitCount()default Long.MAX_VALUE;

    /**
     * 时间段
     */
    longtimeRange();

    /**
     * 阻塞时间段
     */
    longblockDuration();

    /**
     * 时间单位，默认为秒
     */
    TimeUnit timeUnit()default TimeUnit.SECONDS;
}

基于注解的方式限流使用代码如下：

@RateLimit(limitKey = "#key", limitCount = 5, timeRange = 2, blockDuration = 3, timeUnit = TimeUnit.MINUTES)
public String testLimit2(String key){
    ..........
    return key;
}

任何方法添加上述注解具备了一定的限流能力(具体方法需要在 spring aop 指定拦截范围内)，如上代码表示以参数 key 作为限流 key ，每 2 分钟请求次数不超过 5 次，超过限制后阻塞 3 分钟。

分布式锁

• 互斥性:同本地锁一样具有互斥性，但是分布式锁需要保证在不同节点进程的不同线程的互斥。

• 可重入性:同一个节点上的同一个线程如果获取了锁之后那么也可以再次获取这个锁。

• 锁超时:和本地锁一样支持锁超时，防止死锁，通过异步心跳 demon 线程刷新过期时间，防止特殊场景(如 FGC 死锁超时)下死锁。

• 高性能、高可用:加锁和解锁需要高性能，同时也需要保证高可用防止分布式锁失效，可以增加降级。

• 支持阻塞和非阻塞:同 ReentrantLock 一样支持 lock 和 trylock 以及 tryLock ( long timeOut )。

• 公平锁和非公平锁(不支持):公平锁是按照请求加锁的顺序获得锁，非公平锁就相反是无序的，目前 distributed-tools 组件提供的分布式锁不支持该特性。

 /**
     * 分布式锁处理模板执行器
     *
     * @param lockKey 分布式锁key
     * @param resultSupplier 分布式锁处理回调
     * @param waitTime 锁等待时间
     * @param unit 时间单位
     * @param errCodeEnum 指定特殊错误码返回
     * @return
     */
    public static <T> T execute(String lockKey, Supplier<T> resultSupplier, long waitTime, TimeUnit unit,
                                ErrCodeEnum errCodeEnum){
        AssertUtils.assertTrue(StringUtils.isNotBlank(lockKey), ExceptionEnumType.PARAMETER_ILLEGALL);
        boolean locked = false;
        Lock lock = DistributedReentrantLock.newLock(lockKey);
        try {
            locked = waitTime > 0 ? lock.tryLock(waitTime, unit) : lock.tryLock();
        } catch (InterruptedException e) {
            throw new RuntimeException(String.format("lock error,lockResource:%s", lockKey), e);
        }
        if (errCodeEnum != null) {
            AssertUtils.assertTrue(locked, errCodeEnum);
        } else {
            AssertUtils.assertTrue(locked, ExceptionEnumType.ACQUIRE_LOCK_FAIL);
        }
        try {
            return resultSupplier.get();
        } finally {
            lock.unlock();
        }
    }

幂等

幂等 key 提取能力：获取唯一幂等 key

分布式锁服务能力：提供全局加锁、解锁的能力

distributed-tools 幂等组件需要使用自身提供的分布式锁功能，保证其并发唯一性， distributed-tools 提供的分布式锁能够提供其可靠、稳定的加锁、解锁能力。

高性能的写入、查询能力：针对幂等结果查询与存储

distributed-tools 幂等组件提供了基于 tair 、 redis 的存储实现，同时支持自定义一级、二级存储通过 spring 依赖注入到 IdempotentService ，建议 distributed-tools 幂等存储结果一级存储 tair mdb ，二级存储ldb或者 tablestore ，一级存储保证其高性能，二级存储保证其可靠性。

1. 首先判断 Idempotent 的 spelKey 的属性是否为空，如果不为空会根据 spelKey 定义的 spring 表达式生成幂等 ID 。
2. 其次判断参数是否包含 IdempotentTxId 注解，如果有 IdempotentTxId ，会直接获取参数值生成幂等 ID 。
3. 再次通过反射获取参数对象属性是否包含 IdempotentTxId 注解，如果对象属性包含 IdempotentTxId 注解会获取该参数对象属性生成幂等 ID 。
4. 最后以上三种情况仍未获取到幂等 ID ，会进一步通过反射获取参数对象的 Method 是否定义 IdempotentTxIdGetter 注解，如果包含该注解则通过反射生成幂等 ID 。

    @Idempotent(spelKey = "#request.requestId", firstLevelExpireDate = 7，secondLevelExpireDate = 30)
    publicvoidexecute(BizFlowRequest request){
       ..................
    }

/**
     * 幂等模板处理器
     *
     * @param request 幂等Request信息
     * @param executeSupplier 幂等处理回调function
     * @param resultPreprocessConsumer 幂等结果回调function 可以对结果做些预处理
     * @param ifResultNeedIdempotence 除了根据异常还需要根据结果判定是否需要幂等性的场景可以提供此参数
     * @return
     */
    public R execute(IdempotentRequest<P> request, Supplier<R> executeSupplier,
                     Consumer<IdempotentResult<P, R>> resultPreprocessConsumer, Predicate<R> ifResultNeedIdempotence){

      ........
    }

作者信息：

孔凡勇，花名云狄，阿里云-开放平台高级技术家，对高并发、高性能、高可用、可伸缩的分布式系统架构设计有丰富经验，Cloud Native坚定拥护者，坚守开发一线打磨匠艺的架构师。