20180810

限流

public class TokenBucket {// 默認桶大小個數 即最大瞬間流量是64Mprivate static final int DEFAULT_BUCKET_SIZE = 1024 * 1024 * 64;// 一個桶的單位是1字節private int everyTokenSize = 1;// 瞬間最大流量private int maxFlowRate;// 平均流量private int avgFlowRate;// 隊列來緩存桶數量：最大的流量峰值就是 = everyTokenSize*DEFAULT_BUCKET_SIZE 64M = 1 * 1024 *// 1024 * 64private ArrayBlockingQueue<Byte> tokenQueue = new ArrayBlockingQueue<Byte>(DEFAULT_BUCKET_SIZE);private ScheduledExecutorService scheduledExecutorService = Executors.newSingleThreadScheduledExecutor();private volatile boolean isStart = false;private ReentrantLock lock = new ReentrantLock(true);private static final byte A_CHAR = 'a';public TokenBucket() {}public TokenBucket(int maxFlowRate, int avgFlowRate) {this.maxFlowRate = maxFlowRate;this.avgFlowRate = avgFlowRate;}public TokenBucket(int everyTokenSize, int maxFlowRate, int avgFlowRate) {this.everyTokenSize = everyTokenSize;this.maxFlowRate = maxFlowRate;this.avgFlowRate = avgFlowRate;}public void addTokens(Integer tokenNum) {// 若是桶已經滿了，就不再家如新的令牌for (int i = 0; i < tokenNum; i++) {tokenQueue.offer(Byte.valueOf(A_CHAR));}}public TokenBucket build() {start();return this;}/*** 獲取足夠的令牌個數* * @return*/public boolean getTokens(byte[] dataSize) {// Preconditions.checkNotNull(dataSize); // Preconditions.checkArgument(isStart, // "please invoke start method first !");int needTokenNum = dataSize.length / everyTokenSize + 1;// 傳輸內容大小對應的桶個數final ReentrantLock lock = this.lock;lock.lock();try {boolean result = needTokenNum <= tokenQueue.size(); // 是否存在足夠的桶數量if (!result) {return false;}int tokenCount = 0;for (int i = 0; i < needTokenNum; i++) {Byte poll = tokenQueue.poll();if (poll != null) {tokenCount++;}}return tokenCount == needTokenNum;} finally {lock.unlock();}}public void start() {// 初始化桶隊列大小if (maxFlowRate != 0) {tokenQueue = new ArrayBlockingQueue<Byte>(maxFlowRate);}// 初始化令牌生產者TokenProducer tokenProducer = new TokenProducer(avgFlowRate, this);scheduledExecutorService.scheduleAtFixedRate(tokenProducer, 0, 1,TimeUnit.SECONDS);isStart = true;}public void stop() {isStart = false;scheduledExecutorService.shutdown();}public boolean isStarted() {return isStart;}class TokenProducer implements Runnable {private int avgFlowRate;private TokenBucket tokenBucket;public TokenProducer(int avgFlowRate, TokenBucket tokenBucket) {this.avgFlowRate = avgFlowRate;this.tokenBucket = tokenBucket;}@Overridepublic void run() {tokenBucket.addTokens(avgFlowRate);}}public static TokenBucket newBuilder() {return new TokenBucket();}public TokenBucket everyTokenSize(int everyTokenSize) {this.everyTokenSize = everyTokenSize;return this;}public TokenBucket maxFlowRate(int maxFlowRate) {this.maxFlowRate = maxFlowRate;return this;}public TokenBucket avgFlowRate(int avgFlowRate) {this.avgFlowRate = avgFlowRate;return this;}private String stringCopy(String data, int copyNum) {StringBuilder sbuilder = new StringBuilder(data.length() * copyNum);for (int i = 0; i < copyNum; i++) {sbuilder.append(data);}return sbuilder.toString();}public static void main(String[] args) throws IOException,InterruptedException { // arrayTest();tokenTest();}private static void arrayTest() {ArrayBlockingQueue<Integer> tokenQueue = new ArrayBlockingQueue<Integer>(10);tokenQueue.offer(1);tokenQueue.offer(1);tokenQueue.offer(1);System.out.println(tokenQueue.size());System.out.println(tokenQueue.remainingCapacity());}private static void tokenTest() throws InterruptedException, IOException {TokenBucket tokenBucket = TokenBucket.newBuilder().avgFlowRate(512).maxFlowRate(1024).build();BufferedWriter bufferedWriter = new BufferedWriter(new OutputStreamWriter(new FileOutputStream("D:/ds_test")));String data = "xxxx";// 四個字節for (int i = 1; i <= 1000; i++) {Random random = new Random();int i1 = random.nextInt(100);boolean tokens = tokenBucket.getTokens(tokenBucket.stringCopy(data,i1).getBytes());TimeUnit.MILLISECONDS.sleep(100);if (tokens) {bufferedWriter.write("token pass --- index:" + i1);System.out.println("token pass --- index:" + i1);} else {bufferedWriter.write("token rejuect --- index" + i1);System.out.println("token rejuect --- index" + i1);}bufferedWriter.newLine();bufferedWriter.flush();}bufferedWriter.close();}} public class FunnelRateLimiter {static class Funnel {int capacity;float leakingRate;int leftQuota;long leakingTs;public Funnel(int capacity, float leakingRate) {this.capacity = capacity;this.leakingRate = leakingRate;this.leftQuota = capacity;this.leakingTs = System.currentTimeMillis();}void makeSpace() {long nowTs = System.currentTimeMillis();long deltaTs = nowTs - leakingTs;int deltaQuota = (int) (deltaTs * leakingRate);if (deltaQuota < 0) { // 間隔時間太長，整數數字過大溢出this.leftQuota = capacity;this.leakingTs = nowTs;return;}if (deltaQuota < 1) { // 騰出空間太小，最小單位是1return;}this.leftQuota += deltaQuota;this.leakingTs = nowTs;if (this.leftQuota > this.capacity) {this.leftQuota = this.capacity;}}boolean watering(int quota) {makeSpace();if (this.leftQuota >= quota) {this.leftQuota -= quota;return true;}return false;}}private Map<String, Funnel> funnels = new HashMap<String, Funnel>();public boolean isActionAllowed(String userId, String actionKey, int capacity, float leakingRate) {String key = String.format("%s:%s", userId, actionKey);Funnel funnel = funnels.get(key);if (funnel == null) {funnel = new Funnel(capacity, leakingRate);funnels.put(key, funnel);}return funnel.watering(1); // 需要1個quota} }

總結

以上是生活随笔為你收集整理的api限流的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。