前言
在工作中是否会碰到这样的场景，高并发的请求但是无法全部执行，需要一定的限流。如果你是使用的微服务框架，比如SpringCloud，可以使用Gateway增加限流策略来解决。本篇文章是在没有框架的情况实现限流器。
Maven依赖
<dependency> <groupId>com.google.guava</groupId> <artifactId>guava</artifactId> <version>31.0.1-jre</version> </dependency>
代码
上代码，不废话。
首先是限流器代码。
package com.huyi.csdn.tools.rate; import com.google.common.util.concurrent.Monitor; import com.google.common.util.concurrent.RateLimiter; import java.util.concurrent.ConcurrentLinkedQueue; import java.util.function.Consumer; /** * @Program: csdn @ClassName: RequestRateLimiter @Author: huyi @Date: 2021-10-30 22:16 @Description: * 请求限流器 @Version: V1.0 */ public class RequestRateLimiter { // 请求队列 private final ConcurrentLinkedQueue<Request> bucket = new ConcurrentLinkedQueue<>(); // 队列上限 private static final int BUCKET_CAPACITY = 100; // 漏桶下沿水流速度 private final RateLimiter rateLimiter = RateLimiter.create(10.0D); // 请求监视器 private final Monitor requestMoniter = new Monitor(); // 处理监视器 private final Monitor handleMoniter = new Monitor(); /** 请求实体 */ public static class Request { private int data; public Request(int data) { this.data = data; } @Override public String toString() { return "Request{" + "data=" + data + '}'; } } public void submitRequest(int data) { this.submitRequest(new Request(data)); } public void submitRequest(Request request) { // 请求监视器，创建监视向导，队列数据量小于上限 if (requestMoniter.enterIf(requestMoniter.newGuard(() -> bucket.size() < BUCKET_CAPACITY))) { try { boolean result = bucket.offer(request); if (result) { System.out.println("成功向队列加入新的请求！" + Thread.currentThread() + " request:" + request); } else { System.out.println("加入新请求失败！"); } } finally { requestMoniter.leave(); } } else { // 队列已满 // System.out.println("请求队列已经上限，请稍后重试！"); } } // 处理请求方法 public void handleRequest(Consumer<Request> consumer) { if (handleMoniter.enterIf(handleMoniter.newGuard(() -> !bucket.isEmpty()))) { try { // 匀速处理 rateLimiter.acquire(); consumer.accept(bucket.poll()); } finally { handleMoniter.leave(); } } } }
代码说明
1、有个长度100的任务队列，增加了监视器。
2、添加了限流器，限流为10。
验证代码
package com.huyi.csdn.tools.rate; import java.time.LocalTime; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicInteger; import java.util.stream.IntStream; /** * @Program: csdn @ClassName: TestRateLimiter @Author: huyi @Date: 2021-10-30 * 22:35 @Description: @Version: V1.0 */ public class TestRateLimiter { private static final AtomicInteger DATA = new AtomicInteger(0); private static final RequestRateLimiter HANDLE = new RequestRateLimiter(); public static void main(String[] args) { IntStream.range(0, 10) .forEach( (x) -> new Thread( () -> { while (true) { HANDLE.submitRequest(DATA.getAndIncrement()); try { TimeUnit.MILLISECONDS.sleep(100); } catch (InterruptedException e) { e.printStackTrace(); } } }) .start()); IntStream.range(0, 20) .forEach( (x) -> new Thread( () -> { while (true) { HANDLE.handleRequest( y -> System.out.println( LocalTime.now() + ":处理数据 -> " + y.toString())); } }) .start()); } }
验证执行结果
总结
胸中一点浩然气，天地千里快哉风。
如果本文对你有用的，请不要吝啬你的赞，谢谢支持。
