前言

笔者在一家规模还算可以的互联网公司工作，参与优惠券、红包等营销领域的系统的开发，这些系统都是和用户利益密切相关，贴近用户的需求，故备受用户关注，流量极高，俗称“高并发”。如何确保系统的平稳？这就涉及到一个电商公司必备的利器——限流组件，下面来详细描述下实现原理。

实现

如何实现一个限流方法呢，并且是秒级的呢？听起来挺高大上的，其实没那么复杂，使用本地缓存guava，采用计数器算法，写少量代码，即可做到！

package com.example.limit.utils;

import com.google.common.cache.CacheBuilder;
import com.google.common.cache.CacheLoader;
import com.google.common.cache.LoadingCache;

import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * 计数器算法实现限流
 */
public class LimitUtils {

    private final static LoadingCache<String, AtomicInteger> limiter = CacheBuilder
            .newBuilder()
            .maximumSize(Integer.MAX_VALUE)
            .expireAfterWrite(1, TimeUnit.SECONDS)
            .build(new CacheLoader<String, AtomicInteger>() {
                @Override
                public AtomicInteger load(String key) {
                    return new AtomicInteger(0);
                }
            });

    /**
     * 是否被限流
     *
     * @param channelName 渠道
     * @param limitSize   限额
     * @return
     * @throws ExecutionException
     */
    public static boolean isLimited(String channelName, Integer limitSize) throws ExecutionException {
        return limiter.get(channelName).incrementAndGet() > limitSize;
    }

}

以上，展示了计数器算法实现限流的方法，要点有以下几点：

1.计数采用原子整数类AtomicInteger

为了确保限流准确无误，我们不能采用简单的int类型，需要采用原子操作类AtomicInteger，它是一个原子操作，在高并发多线程情况下，能准确计数，不会出现重复计数的等情况。

2.使用本地缓存LoadingCache

限流计算，在实际场景中，qps是十分高的，如何确保性能，只能采用本地缓存，不能采用外置的Redis等。不过有个缺陷，就是本地缓存不能共享计数值，故只能单机限流，每台机器各自限流各自的，比较依赖rpc的负载均衡。

测试

下面我们写个Service，利用这个限流的utils，在多线程下做下测试，看是否能正确限流

首先是定义个接口方法

package com.example.limit.demo;

import java.util.concurrent.ExecutionException;

public interface LimitService {

    /**
     * 测试方法
     *
     * @param channelName 渠道
     * @param limitSize   限额
     * @return
     * @throws ExecutionException
     */
    String doSomething(String channelName, Integer limitSize) throws ExecutionException;

}

接着是接口方法的实现类

package com.example.limit.demo;

import com.example.limit.utils.LimitUtils;
import org.springframework.stereotype.Service;

import java.util.concurrent.ExecutionException;

@Service
public class LimitServiceImpl implements LimitService {

    @Override
    public String doSomething(String channelName, Integer limitSize) throws ExecutionException {
        if (LimitUtils.isLimited(channelName, limitSize)) {
            return "调用失败";
        } else {
            return "调用成功---";
        }
    }
}

最后是多线程情况下测试

package com.example.limit;

import com.example.limit.demo.LimitService;
import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;

import java.io.IOException;
import java.util.concurrent.ExecutionException;

@SpringBootTest
class LimitApplicationTests {

    @Autowired
    LimitService limitService;

    @Test
    void contextLoads() throws IOException {
        exec("线程1", "a", 3);
        exec("线程2", "a", 3);
        // 等待线程执行完
        System.in.read();
    }

    void exec(String threadName, String channelName, Integer limitSize) {
        new Thread(() -> {
            for (int i = 0; i < 10; i++) {
                try {
                    System.out.println(Thread.currentThread().getName() + limitService.doSomething(channelName, limitSize));
                    if (i == 5) {
                        Thread.sleep(1000);
                    }
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        }, threadName).start();
    }

}

结果分析

1.在限流方面，我指定方法的限流为3/s。

2.在线程方面，我发起2个线程，每个线程发起10次请求，共20次

3.在时间方面，我确保每个线程执行的时间超过1秒。

结果可以看到，2秒内，20次请求，只有6次请求成功。

这是符合我们的限流预定的！

其他

本文采用了Guava+AtomicInteger，实现一个计数器限流算法，虽然简单，但在实际场景中，还是经常用到！下次和他人谈起限流方法，可以刚起来了！