feat:单机限流类

mallchat-chat-server/src/main/resources/application.yml

@@ -1,87 +1,87 @@
 logging:
-    level:
+  level:
-        org.springframework.web: INFO
+    org.springframework.web: INFO
-        com.github.binarywang.demo.wx.mp: DEBUG
+    com.github.binarywang.demo.wx.mp: DEBUG
-        me.chanjar.weixin: DEBUG
+    me.chanjar.weixin: DEBUG
 mybatis-plus:
-    mapper-locations: classpath:mapper/**/*.xml
+  mapper-locations: classpath:mapper/**/*.xml
-    configuration:
+  configuration:
-        log-impl: org.apache.ibatis.logging.stdout.StdOutImpl
+    log-impl: org.apache.ibatis.logging.stdout.StdOutImpl
-        map-underscore-to-camel-case: true
+    map-underscore-to-camel-case: true
 spring:
-    profiles:
+  profiles:
-        #运行的环境
+    #运行的环境
-        active: my-test
+    active: my-test
-    application:
+  application:
-        name: mallchat
+    name: mallchat
-    datasource:
+  datasource:
-        url: jdbc:mysql://${mallchat.mysql.ip}:${mallchat.mysql.port}/${mallchat.mysql.db}?useUnicode=true&characterEncoding=UTF-8&serverTimezone=Asia/Shanghai
+    url: jdbc:mysql://${mallchat.mysql.ip}:${mallchat.mysql.port}/${mallchat.mysql.db}?useUnicode=true&characterEncoding=UTF-8&serverTimezone=Asia/Shanghai
-        username: ${mallchat.mysql.username}
+    username: ${mallchat.mysql.username}
-        password: ${mallchat.mysql.password}
+    password: ${mallchat.mysql.password}
-        driver-class-name: com.mysql.cj.jdbc.Driver
+    driver-class-name: com.mysql.cj.jdbc.Driver
-        hikari:
+    hikari:
-            minimum-idle: 3
+      minimum-idle: 3
-            maximum-pool-size: 10
+      maximum-pool-size: 10
-            max-lifetime: 30000   #不能小于30秒，否则默认回到1800秒
+      max-lifetime: 30000   #不能小于30秒，否则默认回到1800秒
-            connection-test-query: SELECT 1
+      connection-test-query: SELECT 1
-    mvc:
+  mvc:
-        pathmatch:
+    pathmatch:
-            matching-strategy: ANT_PATH_MATCHER
+      matching-strategy: ANT_PATH_MATCHER
-    redis:
+  redis:
-        # Redis服务器地址
+    # Redis服务器地址
-        host: ${mallchat.redis.host}
+    host: ${mallchat.redis.host}
-        # Redis服务器端口号
+    # Redis服务器端口号
-        port: ${mallchat.redis.port}
+    port: ${mallchat.redis.port}
-        # 使用的数据库索引，默认是0
+    # 使用的数据库索引，默认是0
-        database: 0
+    database: 0
-        # 连接超时时间
+    # 连接超时时间
-        timeout: 1800000
+    timeout: 1800000
-        # 设置密码
+    # 设置密码
-        password: ${mallchat.redis.password}
+    password: ${mallchat.redis.password}
-    jackson:
+  jackson:
-        serialization:
+    serialization:
-            write-dates-as-timestamps: true
+      write-dates-as-timestamps: true
 jwt:
-    secret: ${mallchat.jwt.secret}
+  secret: ${mallchat.jwt.secret}
 wx:
-    mp:
+  mp:
-        #    callback: http://f4cd-113-92-129-127.ngrok.io
+    #    callback: http://f4cd-113-92-129-127.ngrok.io
-        callback: ${mallchat.wx.callback}
+    callback: ${mallchat.wx.callback}
-        configs:
+    configs:
-            - appId: ${mallchat.wx.appId} # 第一个公众号的appid
+      - appId: ${mallchat.wx.appId} # 第一个公众号的appid
-              secret: ${mallchat.wx.secret} # 公众号的appsecret
+        secret: ${mallchat.wx.secret} # 公众号的appsecret
-              token: ${mallchat.wx.token} # 接口配置里的Token值
+        token: ${mallchat.wx.token} # 接口配置里的Token值
-              aesKey: ${mallchat.wx.aesKey} # 接口配置里的EncodingAESKey值
+        aesKey: ${mallchat.wx.aesKey} # 接口配置里的EncodingAESKey值
 chatai:
-    chatgpt:
+  chatgpt:
-        use: ${mallchat.chatgpt.use}
+    use: ${mallchat.chatgpt.use}
-        AIUserId: ${mallchat.chatgpt.uid}
+    AIUserId: ${mallchat.chatgpt.uid}
-        key: ${mallchat.chatgpt.key}
+    key: ${mallchat.chatgpt.key}
-        proxyUrl: ${mallchat.chatgpt.proxyUrl}
+    proxyUrl: ${mallchat.chatgpt.proxyUrl}
-    chatglm2:
+  chatglm2:
-        use: ${mallchat.chatglm2.use}
+    use: ${mallchat.chatglm2.use}
-        url: ${mallchat.chatglm2.url}
+    url: ${mallchat.chatglm2.url}
-        minute: 3 # 每个用户每3分钟可以请求一次
+    minute: 3 # 每个用户每3分钟可以请求一次
-        AIUserId: ${mallchat.chatglm2.uid}
+    AIUserId: ${mallchat.chatglm2.uid}
 rocketmq:
-    name-server: ${rocketmq.name-server}
+  name-server: ${rocketmq.name-server}
-    # 默认的消息组
+  # 默认的消息组
-    producer:
+  producer:
-        group: chatGroup
+    group: chatGroup
-        send-message-timeout: 3000 # 发送消息超时时间，单位：毫秒。默认为 3000 。
+    send-message-timeout: 3000 # 发送消息超时时间，单位：毫秒。默认为 3000 。
-        compress-message-body-threshold: 4096 # 消息压缩阀值，当消息体的大小超过该阀值后，进行消息压缩。默认为 4 * 1024B
+    compress-message-body-threshold: 4096 # 消息压缩阀值，当消息体的大小超过该阀值后，进行消息压缩。默认为 4 * 1024B
-        max-message-size: 4194304 # 消息体的最大允许大小。。默认为 4 * 1024 * 1024B
+    max-message-size: 4194304 # 消息体的最大允许大小。。默认为 4 * 1024 * 1024B
-        retry-times-when-send-failed: 2 # 同步发送消息时，失败重试次数。默认为 2 次。
+    retry-times-when-send-failed: 2 # 同步发送消息时，失败重试次数。默认为 2 次。
-        retry-times-when-send-async-failed: 2 # 异步发送消息时，失败重试次数。默认为 2 次。
+    retry-times-when-send-async-failed: 2 # 异步发送消息时，失败重试次数。默认为 2 次。
-        retry-next-server: false # 发送消息给 Broker 时，如果发送失败，是否重试另外一台 Broker 。默认为 false
+    retry-next-server: false # 发送消息给 Broker 时，如果发送失败，是否重试另外一台 Broker 。默认为 false
-        access-key: ${rocketmq.access-key} # Access Key ，可阅读 https://github.com/apache/rocketmq/blob/master/docs/cn/acl/user_guide.md 文档
+    access-key: ${rocketmq.access-key} # Access Key ，可阅读 https://github.com/apache/rocketmq/blob/master/docs/cn/acl/user_guide.md 文档
-        secret-key: ${rocketmq.secret-key} # Secret Key
+    secret-key: ${rocketmq.secret-key} # Secret Key
-        enable-msg-trace: true # 是否开启消息轨迹功能。默认为 true 开启。可阅读 https://github.com/apache/rocketmq/blob/master/docs/cn/msg_trace/user_guide.md 文档
+    enable-msg-trace: true # 是否开启消息轨迹功能。默认为 true 开启。可阅读 https://github.com/apache/rocketmq/blob/master/docs/cn/msg_trace/user_guide.md 文档
-        customized-trace-topic: RMQ_SYS_TRACE_TOPIC # 自定义消息轨迹的 Topic 。默认为 RMQ_SYS_TRACE_TOPIC 。
+    customized-trace-topic: RMQ_SYS_TRACE_TOPIC # 自定义消息轨迹的 Topic 。默认为 RMQ_SYS_TRACE_TOPIC 。
-    # Consumer 配置项
+  # Consumer 配置项
-    consumer:
+  consumer:
-        access-key: ${rocketmq.access-key} # Access Key ，可阅读 https://github.com/apache/rocketmq/blob/master/docs/cn/acl/user_guide.md 文档
+    access-key: ${rocketmq.access-key} # Access Key ，可阅读 https://github.com/apache/rocketmq/blob/master/docs/cn/acl/user_guide.md 文档
-        secret-key: ${rocketmq.secret-key} # Secret Key
+    secret-key: ${rocketmq.secret-key} # Secret Key
-        listeners: # 配置某个消费分组，是否监听指定 Topic 。结构为 Map<消费者分组, <Topic, Boolean>> 。默认情况下，不配置表示监听。
+    listeners: # 配置某个消费分组，是否监听指定 Topic 。结构为 Map<消费者分组, <Topic, Boolean>> 。默认情况下，不配置表示监听。
-            erbadagang-consumer-group:
+      erbadagang-consumer-group:
-                topic1: false # 关闭 test-consumer-group 对 topic1 的监听消费
+        topic1: false # 关闭 test-consumer-group 对 topic1 的监听消费

mallchat-tools/mallchat-frequency-control/src/main/java/com/abin/frequencycontrol/service/frequencycontrol/single/FixWindow.java

@@ -0,0 +1,28 @@
+package com.abin.frequencycontrol.service.frequencycontrol.single;
+
+public class FixWindow {
+    public Integer count;  //当前窗口累计请求数
+    public long lastAcquireTime;//最后一次请求时间
+    public Long windowInMillis; //固定窗口时间区间(毫秒)
+    public Integer maxRequests; // 最大请求限制
+
+    public FixWindow(Long windowInMillis, Integer maxRequests) {
+        this.windowInMillis = windowInMillis;
+        this.maxRequests = maxRequests;
+    }
+
+    public synchronized boolean tryAcquire() {
+        long currentTime = System.currentTimeMillis();  //获取系统当前时间
+        //当前和上次不在同一时间窗口
+        if (currentTime - lastAcquireTime > windowInMillis) {
+            count = 0;  // 计数器清0
+            lastAcquireTime = currentTime;  //开启新的时间窗口
+        } else { //同一窗口内
+            if (count < maxRequests) {  // 小于阀值
+                count++;  //计数统计器加1
+                return true;
+            }
+        }
+        return false;
+    }
+}

mallchat-tools/mallchat-frequency-control/src/main/java/com/abin/frequencycontrol/service/frequencycontrol/single/LeakyBucketRateLimiter.java

@@ -0,0 +1,39 @@
+package com.abin.frequencycontrol.service.frequencycontrol.single;
+
+public class LeakyBucketRateLimiter {
+    private final int capacity; // 桶的容量
+    private final int rate; // 出桶速率 (每秒的请求数)
+    private long lastRequestTime; // 上一个请求的时间戳
+
+    public LeakyBucketRateLimiter(int capacity, int rate) {
+        this.capacity = capacity;
+        this.rate = rate;
+        this.lastRequestTime = System.currentTimeMillis();
+    }
+
+    /**
+     * 尝试请求
+     *
+     * @return >0 请求进入桶里，返回的是需要休眠的时间。
+     */
+    public synchronized long tryAcquire() {
+        //当前时间
+        long currentTime = System.currentTimeMillis();
+        //漏桶空的
+        if (currentTime > lastRequestTime) {
+            lastRequestTime = currentTime;
+            return 0; // 请求被允许
+        }
+        //上次取水的间隔时间
+        long elapsedTime = lastRequestTime - currentTime;
+        // 计算桶中的水量
+        int water = (int) elapsedTime * rate / 1000;
+        //水量不超过容量
+        if (water < capacity) {
+            long sleepTime = (1000 / rate) + elapsedTime;
+            lastRequestTime = currentTime + sleepTime;
+            return sleepTime;
+        }
+        return -1; // 请求被拒绝
+    }
+}

mallchat-tools/mallchat-frequency-control/src/main/java/com/abin/frequencycontrol/service/frequencycontrol/single/SlideWindow.java

@@ -0,0 +1,36 @@
+package com.abin.frequencycontrol.service.frequencycontrol.single;
+
+import java.util.LinkedList;
+
+public class SlideWindow {
+
+    private final int maxRequests;//最大请求
+    private final long windowInMillis;//窗口范围
+    private LinkedList<Long> requestTimestamps;//每个请求的时间戳
+
+    public SlideWindow(int maxRequests, long windowInMillis) {
+        this.maxRequests = maxRequests;
+        this.windowInMillis = windowInMillis;
+        this.requestTimestamps = new LinkedList<>();
+    }
+
+    public synchronized boolean tryAcquire() {
+        long currentTime = System.currentTimeMillis();
+        //清除过期窗口的请求
+        cleanExpiredRequests(currentTime);
+        //统计窗口内的请求数小于总限制
+        if (requestTimestamps.size() < maxRequests) {
+            requestTimestamps.addLast(currentTime);
+            return true;
+        }
+
+        return false;
+    }
+
+    private void cleanExpiredRequests(long currentTime) {
+        //由于是LinkedList，头结点就是最早的请求，判断超出时间窗口就移除，留下的都是窗口内的请求
+        while (!requestTimestamps.isEmpty() && (currentTime - requestTimestamps.getFirst() > windowInMillis)) {
+            requestTimestamps.removeFirst();
+        }
+    }
+}

mallchat-tools/mallchat-frequency-control/src/main/java/com/abin/frequencycontrol/service/frequencycontrol/single/Test.java

@@ -0,0 +1,157 @@
+package com.abin.frequencycontrol.service.frequencycontrol.single;
+
+import lombok.SneakyThrows;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+
+public class Test {
+    private static AtomicInteger pass1 = new AtomicInteger();
+    private static AtomicInteger pass2 = new AtomicInteger();
+    private static AtomicInteger pass3 = new AtomicInteger();
+    private static AtomicInteger pass4 = new AtomicInteger();
+    private static AtomicInteger total = new AtomicInteger();
+    private static List<Integer> list1 = new ArrayList<>();
+    private static List<Integer> list2 = new ArrayList<>();
+    private static List<Integer> list3 = new ArrayList<>();
+    private static List<Integer> list4 = new ArrayList<>();
+    private static List<Integer> listTotal = new ArrayList<>();
+    private static List<Double> x = new ArrayList<>();
+    private static volatile boolean stop = false;
+    private static final Integer QPS = 20;
+    private static final Integer accuracy = 4;
+
+
+    public static void main(String[] args) throws InterruptedException {
+        List<Integer> list = Arrays.asList(2, 2, 5, 20, 20, 20, 10, 5, 30, 30, 30, 2, 2, 2, 2, 2, 2, 2, 20, 20, 10, 10, 20, 10, 10, 5, 5, 5, 5, 2, 6, 4, 5, 7, 30, 30, 20, 20, 10, 20, 10, 20, 10, 20, 30, 20, 20, 10, 20, 10, 20, 10, 20, 30, 20, 20, 10, 20, 3, 3, 3, 20, 30, 20, 20, 10, 20, 10, 20, 10, 20);
+        List<Integer> time = split(list);
+        AtomicInteger index = new AtomicInteger(0);
+        tick();
+        TokenBucketRateLimiter tokenBucketRateLimiter = new TokenBucketRateLimiter(30, QPS);
+        ScheduledExecutorService scheduledExecutorService = Executors.newScheduledThreadPool(2);
+        FixWindow fixWindow = new FixWindow(1000L, QPS);
+        SlideWindow slideWindow = new SlideWindow(QPS, 1000L);
+        LeakyBucketRateLimiter leakyBucketRateLimiter = new LeakyBucketRateLimiter(30, QPS);
+        ExecutorService executorService = Executors.newFixedThreadPool(100);
+        ExecutorService common = Executors.newFixedThreadPool(100);
+        scheduledExecutorService.scheduleAtFixedRate(() -> {
+            int andIncrement = index.getAndIncrement();
+            if (andIncrement < time.size()) {
+                Integer integer = time.get(andIncrement);
+                for (Integer i = 0; i < integer; i++) {
+                    common.execute(() -> {
+                        total.incrementAndGet();
+                    });
+                    common.execute(() -> {
+                        fixWindowLimit(fixWindow);
+                    });
+                    common.execute(() -> {
+                        slideWindowLimit(slideWindow);
+                    });
+                    executorService.execute(() -> {
+                        leakyBucketRateLimit(leakyBucketRateLimiter);
+                    });
+                    common.execute(() -> {
+                        tokenBucketRateLimit(tokenBucketRateLimiter);
+                    });
+                }
+            } else {
+                stop = true;
+            }
+
+        }, 0, 250 / accuracy, TimeUnit.MILLISECONDS);
+        Thread.sleep(100000);
+    }
+
+    private static List<Integer> split(List<Integer> result) {
+        ArrayList<Integer> res = new ArrayList<>(result.size() * 4);
+        for (int i = 0; i < result.size(); i++) {
+            Integer integer = result.get(i);
+            Integer divisor = integer / accuracy;
+            if (divisor == 0) divisor = 1;
+            res.add(Math.min(integer, divisor));
+            integer -= divisor;
+            res.add(Math.min(integer, divisor));
+            integer -= divisor;
+            res.add(Math.min(integer, divisor));
+            integer -= divisor;
+            res.add(Math.min(integer, divisor));
+        }
+        return res;
+    }
+
+    private static void tokenBucketRateLimit(TokenBucketRateLimiter tokenBucketRateLimiter) {
+        if (tokenBucketRateLimiter.tryAcquire()) {
+            pass4.incrementAndGet();
+        }
+    }
+
+    private static void leakyBucketRateLimit(LeakyBucketRateLimiter leakyBucketRateLimiter) {
+        long l = leakyBucketRateLimiter.tryAcquire();
+        if (l == 0) {
+            pass3.incrementAndGet();
+        } else if (l > 0) {
+            try {
+                Thread.sleep(l);
+            } catch (InterruptedException e) {
+                e.printStackTrace();
+            }
+            pass3.incrementAndGet();
+        }
+    }
+
+    private static void slideWindowLimit(SlideWindow fixWindow) {
+        boolean b = fixWindow.tryAcquire();
+        if (b) {
+            pass2.incrementAndGet();
+        }
+    }
+
+    private static void fixWindowLimit(FixWindow fixWindow) {
+        boolean b = fixWindow.tryAcquire();
+        if (b) {
+            pass1.incrementAndGet();
+        }
+    }
+
+    private static void limit(List<Integer> result) {
+    }
+
+    private static void tick() {
+        Thread timer = new Thread(new TimerTask());
+        timer.setName("sentinel-timer-task");
+        timer.start();
+    }
+
+    static class TimerTask implements Runnable {
+        @SneakyThrows
+        @Override
+        public void run() {
+            long start = System.currentTimeMillis();
+            int i = 0;
+            while (!stop) {
+                i++;
+                Thread.sleep(250);
+                list1.add(pass1.getAndSet(0));
+                list2.add(pass2.getAndSet(0));
+                list3.add(pass3.getAndSet(0));
+                list4.add(pass4.getAndSet(0));
+                listTotal.add(total.getAndSet(0));
+                x.add(i * 0.25);
+            }
+            System.out.println(x);
+            System.out.println(listTotal);
+            System.out.println(list1);
+            System.out.println(list2);
+            System.out.println(list3);
+            System.out.println(list4);
+
+        }
+    }
+}

mallchat-tools/mallchat-frequency-control/src/main/java/com/abin/frequencycontrol/service/frequencycontrol/single/TokenBucketRateLimiter.java

@@ -0,0 +1,41 @@
+package com.abin.frequencycontrol.service.frequencycontrol.single;
+
+public class TokenBucketRateLimiter {
+    private final int capacity; // 令牌桶容量
+    private final int rate; // 令牌产生速率 (每秒的令牌数)
+    private int tokens; // 当前令牌数量
+    private long lastRefillTime; // 上次令牌补充时间
+    private long left = 0;
+
+    public TokenBucketRateLimiter(int capacity, int rate) {
+        this.capacity = capacity;
+        this.rate = rate;
+        this.tokens = capacity;
+        this.lastRefillTime = System.currentTimeMillis();
+    }
+
+    public synchronized boolean tryAcquire() {
+        refillTokens();
+        if (tokens > 0) {
+            tokens--;
+            return true;
+        }
+        return false;
+    }
+
+    //补充令牌
+    private void refillTokens() {
+        long currentTime = System.currentTimeMillis();
+        //当前时间和上次请求时间相差的令牌数
+        long elapsedTime = currentTime - lastRefillTime;
+
+        if (elapsedTime > 0) {
+            //需要补充的令牌
+            int newTokens = (int) ((elapsedTime * rate + left) / 1000);
+            left = (elapsedTime * rate + left) % 1000;
+            //令牌不能超过桶的大小
+            tokens = Math.min(tokens + newTokens, capacity);
+            lastRefillTime = currentTime;
+        }
+    }
+}