分布式限流

gistfile1.txt

package com.example.anchnet;

import org.redisson.api.RRateLimiter;
import org.redisson.api.RateIntervalUnit;
import org.redisson.api.RateType;
import org.redisson.api.RedissonClient;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;
import java.time.LocalDateTime;
import java.util.Arrays;
import java.util.List;
import java.util.stream.Collectors;
import java.util.stream.IntStream;

@Service
public class SyncService {

    @Resource
    private RedissonClient redissonClient;

    private static final int AK_COUNT = 70;
    private static final int RATE = 20;
    private static final int RATE_INTERVAL = 1;
    private static final RateIntervalUnit RATE_INTERVAL_UNIT = RateIntervalUnit.SECONDS;
    private static final int SYNC_INVOKE_COUNT = 100;
    private static final String[] METRICS = {"cpu", "mem", "wan", "disk", "lb"};
    private static final List<String> AKS = IntStream.range(0, AK_COUNT).boxed().map(item -> "AK" + item).collect(Collectors.toList());

    private void sync(String ak, String metric) {
        String str = LocalDateTime.now().toLocalTime().toString() + "-----sync-----" + metric + "------" + ak;
        System.out.println(str);
    }

    /**
     * 多线程异步执行
     */
    public void execute() {
        Arrays.stream(METRICS).forEach(metric -> {
            // 一个指标一个线程
            new Thread(() -> {
                // 遍历所有的ak
                AKS.forEach(ak -> {
                    // 根据ak获取限流器
                    RRateLimiter rateLimiter = getRRateLimiter(ak);
                    // 一个ak一个线程
                    new Thread(() -> {
                        // 调用几次
                        IntStream.range(0, SYNC_INVOKE_COUNT).forEach(item -> {
                            rateLimiter.acquire();
                            sync(ak, metric);
                        });
                    }).start();
                });
            }).start();
        });
    }

    /**
     * 单线程同步执行
     */
    public void run() {
        Arrays.stream(METRICS).forEach(metric -> {
            AKS.forEach(ak -> {
                RRateLimiter rateLimiter = getRRateLimiter(ak);
                IntStream.range(0, SYNC_INVOKE_COUNT).forEach(item -> {
                    rateLimiter.acquire();
                    sync(ak, metric);
                });
            });
        });
    }

    /**
     * 根据ak构造限流器
     *
     * @param ak
     * @return
     */
    private RRateLimiter getRRateLimiter(String ak) {
        RRateLimiter rateLimiter = redissonClient.getRateLimiter(ak);
        rateLimiter.setRate(RateType.OVERALL, RATE, RATE_INTERVAL, RATE_INTERVAL_UNIT);
        return rateLimiter;
    }

}