diff --git a/springboot-chat-stream/README.md b/springboot-chat-stream/README.md new file mode 100644 index 0000000..7566bbb --- /dev/null +++ b/springboot-chat-stream/README.md @@ -0,0 +1,355 @@ +# 像 ChatGPT 一样丝滑：Spring Boot 如何实现大模型流式（Streaming）响应？ + +## 一、为什么需要流式响应？ + +同样的 HTTP 请求，为什么像 ChatGPT 这类模型的回答能像打字机一样逐字输出，而我们平时写的接口却要等全部处理完才返回？ + +问题的核心在于 **响应模式**： + +| 传统模式 | 流式模式 | +|---------|---------| +| 服务器处理完成 → 一次性返回 | 生成一部分 → 立即推送 | +| 客户端等待总时长 = 服务器处理时间 | 客户端首字等待时间通常很短 | +| 适合快速查询 | 适合耗时生成 | + +对于大模型这种 **生成式 AI**，一个响应可能需要几秒甚至几十秒。如果用传统模式，用户体验就是： + +``` +提问 → (10秒空白) → 答案全部出现 +``` + +而流式响应的体验是： + +``` +提问 → 0.1秒后 → "我" → "认" → "为" → ... → 逐字呈现 +``` + +实现这种效果有多种技术方案，本文将介绍基于 Spring Boot WebFlux + SSE 的实现方式。 + +--- + +## 二、核心技术选型 + +实现流式响应主要有以下几种方案： + +| 方案 | 优点 | 缺点 | 适用场景 | +|------|------|------|----------| +| **SSE** | 单向推送、HTTP协议、实现简单 | 不支持双向通信 | 服务端主动推送 | +| **WebSocket** | 双向通信、实时性强 | 实现复杂、需要额外协议 | 聊天、游戏 | +| **长轮询** | 兼容性好 | 资源消耗大 | 低频数据更新 | + +**本文选择 SSE 方案**，原因如下： +- Spring Boot 原生支持 `ResponseEntity>` +- 基于标准 HTTP，无需额外协议协商 +- 代码简洁，易于理解和维护 + +--- + +## 三、项目依赖配置 + +### 3.1 Maven 依赖 + +```xml + + + 4.0.0 + + + org.springframework.boot + spring-boot-starter-parent + 3.2.0 + + + + com.example + springboot-chat-stream + 1.0.0 + + + 17 + + + + + + org.springframework.boot + spring-boot-starter-webflux + + + + + org.projectlombok + lombok + true + + + +``` + +### 3.2 关键依赖说明 + +- **spring-boot-starter-webflux**：提供响应式 Web 支持，核心是 Reactor 的 `Flux` 类型 +- **Reactor**：响应式编程库，`Flux` 表示 0-N 个元素的异步序列 + +--- + +## 四、核心代码实现 + +### 4.1 Controller 层：流式响应入口 + +```java +package com.example.chat.controller; + +import com.example.chat.service.StreamChatService; +import lombok.RequiredArgsConstructor; +import org.springframework.http.MediaType; +import org.springframework.http.ResponseEntity; +import org.springframework.web.bind.annotation.*; +import reactor.core.publisher.Flux; + +@RestController +@RequestMapping("/api/chat") +@RequiredArgsConstructor +@CrossOrigin(origins = "*") // 开发环境允许跨域 +public class StreamChatController { + + private final StreamChatService chatService; + + /** + * 流式聊天接口 + * @param prompt 用户输入的问题 + * @return 流式响应，text/event-stream 格式 + */ + @GetMapping(value = "/stream", produces = MediaType.TEXT_EVENT_STREAM_VALUE) + public ResponseEntity> streamChat(@RequestParam String prompt) { + return ResponseEntity.ok() + .header("Cache-Control", "no-cache") + .header("Connection", "keep-alive") + .body(chatService.streamResponse(prompt)); + } +} +``` + +**关键点解析：** + +1. `produces = MediaType.TEXT_EVENT_STREAM_VALUE`：声明返回 SSE 格式 +2. `Flux`：响应式流，可以发送多个数据块 +3. `Cache-Control: no-cache`：禁用缓存，确保实时推送 + +### 4.2 Service 层：模拟大模型流式生成 + +```java +package com.example.chat.service; + +import lombok.extern.slf4j.Slf4j; +import org.springframework.stereotype.Service; +import reactor.core.publisher.Flux; + +import java.time.Duration; + +@Slf4j +@Service +public class StreamChatService { + + /** + * 模拟大模型流式生成响应 + * @param prompt 用户问题 + * @return 按字符/词汇流式输出的响应 + */ + public Flux streamResponse(String prompt) { + log.info("收到用户提问: {}", prompt); + + // 模拟大模型生成的回复内容 + String response = mockLLMResponse(prompt); + + // 将响应拆分为字符流，每 50ms 发送一个字符 + return Flux.fromArray(response.split("")) + .delayElements(Duration.ofMillis(50)) + .doOnNext(chunk -> log.debug("发送数据块: {}", chunk)) + .doOnComplete(() -> log.info("流式响应完成")) + .doOnError(e -> log.error("流式响应异常", e)); + } + + /** + * 模拟大模型生成内容（实际项目可接入 OpenAI/通义千问等） + */ + private String mockLLMResponse(String prompt) { + return """ + 【Spring Boot 流式响应】 + 您的问题是：%s + + 这是一个模拟大模型流式输出的示例。 + 在实际应用中，你可以： + 1. 接入 OpenAI API 使用 GPT-4 + 2. 接入阿里云通义千问 API + 3. 接入本地部署的大模型 + + 流式响应的核心是： + - 使用 Spring WebFlux 的 Flux + - 返回 text/event-stream 格式 + - 前端使用 EventSource 或 fetch 接收 + + 这样就能实现像 ChatGPT 一样的丝滑体验！ + """.formatted(prompt); + } +} +``` + +**核心逻辑：** + +1. `Flux.fromArray(response.split(""))`：将字符串拆分为字符数组转为流 +2. `.delayElements(Duration.ofMillis(50))`：每个字符延迟 50ms 发送 +3. `.doOnNext()/.doOnComplete()/.doOnError()`：生命周期钩子，用于日志记录 + +### 4.3 接入真实大模型 API（扩展） + +```java +// 接入 OpenAI Streaming API 的示例（伪代码） +public Flux streamOpenAI(String prompt) { + WebClient webClient = WebClient.builder() + .baseUrl("https://api.openai.com/v1") + .defaultHeader(HttpHeaders.AUTHORIZATION, "Bearer YOUR_API_KEY") + .build(); + + return webClient.post() + .uri("/chat/completions") + .bodyValue(Map.of( + "model", "gpt-4", + "messages", List.of(Map.of("role", "user", "content", prompt)), + "stream", true + )) + .retrieve() + .bodyToFlux(String.class) + .map(this::extractContentFromSSE); // 解析 SSE 格式提取 content +} +``` + +### 4.4 启动类 + +```java +package com.example.chat; + +import org.springframework.boot.SpringApplication; +import org.springframework.boot.autoconfigure.SpringBootApplication; + +@SpringBootApplication +public class ChatStreamApplication { + public static void main(String[] args) { + SpringApplication.run(ChatStreamApplication.class, args); + } +} +``` + +### 4.5 配置文件 + +```yaml +server: + port: 8080 + +spring: + application: + name: chat-stream-demo + +# 日志配置 +logging: + level: + com.example.chat: DEBUG +``` + +--- + +## 五、前端对接示例 + +### 5.1 使用 EventSource 接收流 + +```html + + + + + Spring Boot 流式响应示例 + + + +

Spring Boot 流式聊天

+ + +
+ + + + +``` + +### 5.2 使用 Fetch API（推荐） + +```javascript +async function streamChat(prompt) { + const response = await fetch(`/api/chat/stream?prompt=${encodeURIComponent(prompt)}`); + const reader = response.body.getReader(); + const decoder = new TextDecoder(); + + while (true) { + const { done, value } = await reader.read(); + if (done) break; + + const chunk = decoder.decode(value); + console.log('收到数据:', chunk); + // 更新 UI + } +} +``` + +--- + +## 六、运行效果 + +启动项目后，访问 `http://localhost:8080`（需添加静态页面支持），输入问题，你会看到： + +``` +【Spring Boot 流式响应】 +您的问题是：如何学习 Spring Boot？ + +这是一个模拟大模型流式输出的示例。 +... +``` + +文字像打字机一样逐字出现，体验丝滑！ + + +--- + +## 七、总结 + +本文介绍了如何使用 Spring Boot WebFlux 实现 SSE 流式响应。核心是通过 `Flux` + `TEXT_EVENT_STREAM_VALUE` 将数据分块推送，配合前端 `EventSource` 实现逐字显示效果。相比传统一次性返回，流式响应能显著降低用户等待感知，特别适合大模型对话等耗时生成场景。 diff --git a/springboot-chat-stream/pom.xml b/springboot-chat-stream/pom.xml new file mode 100644 index 0000000..f7fa439 --- /dev/null +++ b/springboot-chat-stream/pom.xml @@ -0,0 +1,48 @@ + + + 4.0.0 + + + org.springframework.boot + spring-boot-starter-parent + 3.2.0 + + + + com.example + springboot-chat-stream + 1.0.0 + Spring Boot Chat Stream Demo + 流式响应演示项目 - 像 ChatGPT 一样丝滑 + + + 17 + + + + + + org.springframework.boot + spring-boot-starter-webflux + + + + + org.projectlombok + lombok + true + + + + + + + org.springframework.boot + spring-boot-maven-plugin + + + + diff --git a/springboot-chat-stream/src/main/java/com/example/chat/ChatStreamApplication.java b/springboot-chat-stream/src/main/java/com/example/chat/ChatStreamApplication.java new file mode 100644 index 0000000..deb5369 --- /dev/null +++ b/springboot-chat-stream/src/main/java/com/example/chat/ChatStreamApplication.java @@ -0,0 +1,15 @@ +package com.example.chat; + +import org.springframework.boot.SpringApplication; +import org.springframework.boot.autoconfigure.SpringBootApplication; + +/** + * Spring Boot 流式响应演示应用 + */ +@SpringBootApplication +public class ChatStreamApplication { + + public static void main(String[] args) { + SpringApplication.run(ChatStreamApplication.class, args); + } +} diff --git a/springboot-chat-stream/src/main/java/com/example/chat/controller/StreamChatController.java b/springboot-chat-stream/src/main/java/com/example/chat/controller/StreamChatController.java new file mode 100644 index 0000000..e245810 --- /dev/null +++ b/springboot-chat-stream/src/main/java/com/example/chat/controller/StreamChatController.java @@ -0,0 +1,36 @@ +package com.example.chat.controller; + +import com.example.chat.service.StreamChatService; +import lombok.RequiredArgsConstructor; +import org.springframework.http.MediaType; +import org.springframework.http.ResponseEntity; +import org.springframework.web.bind.annotation.CrossOrigin; +import org.springframework.web.bind.annotation.GetMapping; +import org.springframework.web.bind.annotation.RequestMapping; +import org.springframework.web.bind.annotation.RequestParam; +import org.springframework.web.bind.annotation.RestController; +import reactor.core.publisher.Flux; + +/** + * 流式聊天控制器 + * 提供 SSE 流式响应接口 + */ +@RestController +@RequestMapping("/api/chat") +@RequiredArgsConstructor +@CrossOrigin(origins = "*") +public class StreamChatController { + + private final StreamChatService chatService; + + /** + * 流式聊天接口（SSE） + * + * @param prompt 用户输入的问题 + * @return SSE 流式响应 + */ + @GetMapping(value = "/stream", produces = MediaType.TEXT_EVENT_STREAM_VALUE) + public Flux streamChat(@RequestParam String prompt) { + return chatService.streamResponse(prompt); + } +} diff --git a/springboot-chat-stream/src/main/java/com/example/chat/service/StreamChatService.java b/springboot-chat-stream/src/main/java/com/example/chat/service/StreamChatService.java new file mode 100644 index 0000000..81f1053 --- /dev/null +++ b/springboot-chat-stream/src/main/java/com/example/chat/service/StreamChatService.java @@ -0,0 +1,77 @@ +package com.example.chat.service; + +import lombok.extern.slf4j.Slf4j; +import org.springframework.stereotype.Service; +import reactor.core.publisher.Flux; + +import java.time.Duration; + +/** + * 流式聊天服务 + * 模拟大模型流式生成响应 + */ +@Slf4j +@Service +public class StreamChatService { + + /** + * 模拟大模型流式生成响应 + * + * @param prompt 用户问题 + * @return 按字符/词汇流式输出的响应 + */ + public Flux streamResponse(String prompt) { + log.info("收到用户提问: {}", prompt); + + // 模拟大模型生成的回复内容 + String response = mockLLMResponse(prompt); + + // 将文本拆分成小块，使用响应式延迟模拟打字效果 + int chunkSize = 2; // 每次发送 2 个字符 + + return Flux.fromArray(splitIntoChunks(response, chunkSize)) + .delayElements(Duration.ofMillis(30)) // 每 30ms 发送一个块 + .doOnComplete(() -> log.info("流式响应完成")); + } + + /** + * 将字符串拆分成固定大小的块 + */ + private String[] splitIntoChunks(String text, int chunkSize) { + int length = (text.length() + chunkSize - 1) / chunkSize; + String[] chunks = new String[length]; + for (int i = 0; i < length; i++) { + int start = i * chunkSize; + int end = Math.min(start + chunkSize, text.length()); + chunks[i] = text.substring(start, end); + } + return chunks; + } + + /** + * 模拟大模型生成内容 + * 实际项目可接入 OpenAI/通义千问等 API + * + * @param prompt 用户问题 + * @return 模拟的回复内容 + */ + private String mockLLMResponse(String prompt) { + return """ + 【Spring Boot 流式响应】 + 您的问题是：%s + + 这是一个模拟大模型流式输出的示例。 + 在实际应用中，你可以： + 1. 接入 OpenAI API 使用 GPT-4 + 2. 接入阿里云通义千问 API + 3. 接入本地部署的大模型 + + 流式响应的核心是： + - 使用 Spring WebFlux 的 Flux + - 返回 text/event-stream 格式 + - 前端使用 EventSource 或 fetch 接收 + + 这样就能实现像 ChatGPT 一样的丝滑体验！ + """.formatted(prompt); + } +} diff --git a/springboot-chat-stream/src/main/resources/application.yml b/springboot-chat-stream/src/main/resources/application.yml new file mode 100644 index 0000000..3c2bc6b --- /dev/null +++ b/springboot-chat-stream/src/main/resources/application.yml @@ -0,0 +1,13 @@ +server: + port: 8080 + +spring: + application: + name: chat-stream-demo + +# 日志配置 +logging: + level: + com.example.chat: DEBUG + pattern: + console: "%d{HH:mm:ss.SSS} [%thread] %-5level %logger{36} - %msg%n" diff --git a/springboot-chat-stream/src/main/resources/static/index.html b/springboot-chat-stream/src/main/resources/static/index.html new file mode 100644 index 0000000..b7ae4f1 --- /dev/null +++ b/springboot-chat-stream/src/main/resources/static/index.html @@ -0,0 +1,667 @@ + + + + + + AI 流式响应演示 + + + + + + + +
+ +
+
+
+ 在线 +
+
+
+ + +
+ +
+

👋 欢迎体验 AI 流式响应

+

基于 Spring Boot WebFlux + SSE 实现实时流式输出
输入任意问题，感受像 ChatGPT 一样的逐字显示效果

+
+ + +
+
+ +
+
🤖
+
+
+ AI 助手 + 现在 +
+
+ 你好！我是基于 Spring Boot 的流式响应助手。你可以问我任何问题，我会逐字回复，让你体验流畅的打字效果！ +
+
+
+
+ + +
+
+
+ +
+ +
+
+
+
+ + + + diff --git a/springboot-netspeed-limit/README.md b/springboot-netspeed-limit/README.md new file mode 100644 index 0000000..a3a12f1 --- /dev/null +++ b/springboot-netspeed-limit/README.md @@ -0,0 +1,287 @@ +# 概述 + +本文介绍在 Spring Boot 3 中实现多维度网络带宽限速的完整方案。基于**令牌桶算法**手动实现核心逻辑，通过自定义 `HandlerInterceptor` 拦截请求、`HttpServletResponseWrapper` 包装响应流、`RateLimitedOutputStream` 控制输出速率，实现对文件下载、视频流等场景的精确速度控制。 + +# 为什么需要带宽限速 + +带宽限速与常见的 API 限流不同：限流控制的是**请求次数**（如每分钟100次），而限速控制的是**网络带宽**（如每秒200KB）。在实际应用中，带宽限速有着重要的业务价值： + +**场景一：文件下载服务** +对于网盘或资源分发平台，免费用户限制在 200KB/s，VIP 用户提升到 2MB/s，既能保障基础体验，又能激励付费转化。 + +**场景二：视频流媒体** +不同清晰度对应不同带宽限制（480P 用 500KB/s，1080P 用 3MB/s），避免高码率视频占用过多服务器带宽。 + +**场景三：API 接口保护** +大数据量接口（如导出报表）如果没有带宽控制，单个请求可能占满整个出口带宽，影响其他用户访问。 + +# 核心原理：令牌桶算法 + +令牌桶算法是流量控制的经典方案，其思想非常直观：想象一个桶，系统以固定速率向桶中放入令牌，请求数据时必须从桶中取走对应数量的令牌。 + +**核心参数解析：** + +**1. 桶容量（Capacity）**：决定能承受多大突发流量。容量为 200KB 时，即使桶已满，最多也只能连续发送 200KB 数据，之后必须等待令牌补充。 + +**2. 填充速率（Refill Rate）**：决定长期平均传输速度。每秒补充 200KB 令牌，意味着平均速度就是 200KB/s。 + +**3. 分块大小（Chunk Size）**：影响流量平滑度。将 8KB 数据拆分成 2KB×4 次写入，每次写入之间进行令牌检查，比一次性写入 8KB 更加平滑。 + +**算法流程：** +``` +发送数据前： +1. 计算距离上次补充的时间差 +2. 根据 时间差 × 填充速率 计算新增令牌数 +3. 更新桶中令牌数（不超过容量上限） + +发送数据时： +1. 检查令牌是否足够 +2. 足够：直接扣除令牌，发送数据 +3. 不足：计算 (缺少令牌数 / 填充速率) 得到等待时间，精确等待后发送 +``` + +# 技术设计 + +### 整体流程 + +本方案采用拦截器模式，在请求处理的早期阶段完成限速组件的初始化，通过请求属性传递包装后的响应对象。 + +``` +请求流程： +┌─────────────────────────────────────────────────────────────────────┐ +│ 1. DispatcherServlet 分发请求 │ +└─────────────────────────────────────────────────────────────────────┘ + ↓ +┌─────────────────────────────────────────────────────────────────────┐ +│ 2. BandwidthLimitInterceptor.preHandle() │ +│ - 解析 @BandwidthLimit 注解 │ +│ - 从 BandwidthLimitManager 获取共享 TokenBucket │ +│ - 创建 BandwidthLimitResponseWrapper 并存入 request attribute │ +└─────────────────────────────────────────────────────────────────────┘ + ↓ +┌─────────────────────────────────────────────────────────────────────┐ +│ 3. Controller 处理请求 │ +│ - 通过 BandwidthLimitHelper.getLimitedResponse() 获取包装后的响应 │ +│ - 向响应流写入数据（自动触发限速） │ +└─────────────────────────────────────────────────────────────────────┘ + ↓ +┌─────────────────────────────────────────────────────────────────────┐ +│ 4. BandwidthLimitInterceptor.afterCompletion() │ +│ - 清理资源，关闭流 │ +└─────────────────────────────────────────────────────────────────────┘ +``` + +### 为什么选择 HandlerInterceptor + +在 Spring Boot 中实现请求处理，有两种常见方式：Filter 和 HandlerInterceptor。本方案选择 HandlerInterceptor 的关键原因是：**注解解析需要 HandlerMethod 对象**。 + +Filter 在 DispatcherServlet 之前执行，此时还没有确定具体的处理方法，无法获取方法上的 `@BandwidthLimit` 注解。而 HandlerInterceptor 在处理器确定后执行，可以通过 `HandlerMethod` 精确获取方法级别和类级别的注解信息。 + +### 核心组件职责 + +| 组件 | 职责 | +|------|------| +| `@BandwidthLimit` | 声明式注解，配置限速参数 | +| `BandwidthLimitInterceptor` | 拦截请求，解析注解，创建响应包装器 | +| `BandwidthLimitManager` | 管理多维度限速桶（全局/API/用户/IP） | +| `BandwidthLimitResponseWrapper` | 包装 HttpServletResponse，替换 OutputStream | +| `RateLimitedOutputStream` | 实现限速逻辑，包装 TokenBucket | +| `TokenBucket` | 令牌桶算法实现 | +| `BandwidthLimitHelper` | 从请求属性中获取包装后的响应对象 | + +# 多维度限速实现 + +本方案支持四种限速维度，满足不同业务场景需求： + +### 全局限速（GLOBAL） + +所有请求共享同一个限速桶，适合保护服务器整体出口带宽。例如设置 10MB/s 全局限制，即使有100个并发下载，总带宽也不会超过 10MB/s。 + +```java +@BandwidthLimit(value = 200, unit = BandwidthUnit.KB, type = LimitType.GLOBAL) +@GetMapping("/download/global") +public void downloadGlobal(HttpServletResponse response) throws IOException { + HttpServletResponse limitedResponse = BandwidthLimitHelper.getLimitedResponse(request, response); + // 写入数据... +} +``` + +### API 维度限速（API） + +每个接口路径独立限速，不同接口的流量互不影响。`/api/file/download` 限制 500KB/s，`/api/video/stream` 限制 2MB/s，两个接口可以同时达到各自的速度上限。 + +```java +@BandwidthLimit(value = 500, unit = BandwidthUnit.KB, type = LimitType.API) +@GetMapping("/download/file") +public void downloadFile(HttpServletResponse response) throws IOException { + // 文件下载逻辑 +} + +@BandwidthLimit(value = 2048, unit = BandwidthUnit.KB, type = LimitType.API) +@GetMapping("/stream/video") +public void streamVideo(HttpServletResponse response) throws IOException { + // 视频流逻辑 +} +``` + +### 用户维度限速（USER） + +根据用户标识（如请求头 `X-User-Id`）进行限速，每个用户独立计算带宽。配合 `free` 和 `vip` 参数，可实现差异化服务： + +```java +@BandwidthLimit(value = 200, unit = BandwidthUnit.KB, type = LimitType.USER, + free = 200, vip = 2048) +@GetMapping("/download/user") +public void downloadByUser(@RequestHeader("X-User-Type") String userType, + HttpServletResponse response) throws IOException { + // 根据请求头 X-User-Type 自动应用 200KB/s 或 2MB/s 限速 +} +``` + +### IP 维度限速（IP） + +根据客户端 IP 地址限速，防止单个 IP 占用过多带宽。支持代理环境下的 IP 获取（X-Forwarded-For、X-Real-IP）。 + +```java +@BandwidthLimit(value = 300, unit = BandwidthUnit.KB, type = LimitType.IP) +@GetMapping("/download/ip") +public void downloadByIp(HttpServletResponse response) throws IOException { + // 每个独立 IP 限制 300KB/s +} +``` + +# 关键代码实现 + +### 1. 令牌桶核心算法 + +TokenBucket 的核心在于精确的时间计算和令牌补充。使用 `System.nanoTime()` 获取纳秒级时间戳，确保高精度速率控制。 + +```java +public synchronized void acquire(long permits) { + // 1. 补充令牌 + refill(); + + // 2. 计算等待时间 + if (tokens >= permits) { + tokens -= permits; + return; + } + + long deficit = permits - tokens; + long waitNanos = (deficit * 1_000_000_000L) / refillRate; + + // 3. 精确等待 + sleepNanos(waitNanos); + + // 4. 等待后消费 + tokens = 0; +} + +private void refill() { + long now = System.nanoTime(); + long elapsedNanos = now - lastRefillTime; + long newTokens = (elapsedNanos * refillRate) / 1_000_000_000L; + tokens = Math.min(capacity, tokens + newTokens); + lastRefillTime = now; +} +``` + +### 2. 响应包装器 + +HttpServletResponseWrapper 是 Servlet 规范提供的响应包装基类，通过覆盖 `getOutputStream()` 方法返回自定义的限速输出流。 + +```java +public class BandwidthLimitResponseWrapper extends HttpServletResponseWrapper { + private final TokenBucket sharedTokenBucket; // 共享的令牌桶 + + @Override + public ServletOutputStream getOutputStream() throws IOException { + if (limitedOutputStream == null && sharedTokenBucket != null) { + // 使用共享 TokenBucket，确保多维度统计正确 + limitedOutputStream = new RateLimitedOutputStream( + super.getOutputStream(), + sharedTokenBucket, + bandwidthBytesPerSecond + ); + } + return limitedOutputStream; + } +} +``` + +### 3. 拦截器获取包装响应 + +拦截器在 `preHandle` 中创建响应包装器，存储到 request attribute，Controller 通过 `BandwidthLimitHelper` 获取。 + +```java +@Override +public boolean preHandle(HttpServletRequest request, HttpServletResponse response, Object handler) { + BandwidthLimit annotation = findAnnotation(handler); + if (annotation != null) { + // 从 Manager 获取共享 TokenBucket + TokenBucket bucket = limitManager.getBucket(type, key, capacity, rate); + + // 创建包装器并存储 + BandwidthLimitResponseWrapper wrappedResponse = + new BandwidthLimitResponseWrapper(response, bucket, bandwidthBytesPerSecond, chunkSize); + request.setAttribute("BandwidthLimitWrappedResponse", wrappedResponse); + } + return true; +} +``` + +### 4. Controller 获取限速响应 + +Controller 通过 `BandwidthLimitHelper.getLimitedResponse()` 获取包装后的响应，所有写入操作都会自动限速。 + +```java +@GetMapping("/download/global") +public void downloadGlobal(HttpServletRequest request, HttpServletResponse response) throws IOException { + HttpServletResponse limitedResponse = BandwidthLimitHelper.getLimitedResponse(request, response); + + limitedResponse.setContentType("application/octet-stream"); + limitedResponse.setHeader("Content-Disposition", "attachment; filename=test.bin"); + + // 写入数据时自动限速 + limitedResponse.getOutputStream().write(data); +} +``` + +# 参数调优指南 + +### 桶容量选择 + +容量决定突发流量承受能力： + +| 容量设置 | 突发能力 | 适用场景 | +|----------|----------|----------| +| 速率 × 0.5 | 平滑，无突发 | 流量控制严格的场景 | +| 速率 × 1.0 | 允许 1 秒突发 | 默认推荐值 | +| 速率 × 2.0 | 允许 2 秒突发 | 需要良好首屏加载 | + +```java +// 注解配置 +@BandwidthLimit(value = 200, unit = BandwidthUnit.KB, capacityMultiplier = 1.0) +``` + +### 分块大小选择 + +分块大小影响流量平滑度，经验公式：`chunkSize = bandwidth / 50` + +| 带宽 | 推荐分块 | 理由 | +|------|----------|------| +| 200 KB/s | 1-4 KB | 小分块保证平滑 | +| 1 MB/s | 4-8 KB | 平衡平滑与性能 | +| 5 MB/s+ | 8-16 KB | 减少系统调用开销 | + +```java +// 自动计算（推荐） +@BandwidthLimit(value = 200, unit = BandwidthUnit.KB, chunkSize = -1) + +// 手动指定 +@BandwidthLimit(value = 200, unit = BandwidthUnit.KB, chunkSize = 4096) +``` + +# 总结 + +本文基于令牌桶算法，通过 HandlerInterceptor + HttpServletResponseWrapper，在 Spring Boot 中实现了多维度带宽限速。支持全局/API/用户/IP 四种限速维度，提供实时统计监控，适用于API接口保护、文件下载、视频流等场景。 \ No newline at end of file diff --git a/springboot-netspeed-limit/pom.xml b/springboot-netspeed-limit/pom.xml new file mode 100644 index 0000000..ee1747c --- /dev/null +++ b/springboot-netspeed-limit/pom.xml @@ -0,0 +1,80 @@ + + + 4.0.0 + + + org.springframework.boot + spring-boot-starter-parent + 3.3.0 + + + + com.example + springboot-netspeed-limit + 1.0.0 + Spring Boot Network Speed Limit + Bandwidth limit implementation with Token Bucket algorithm + + + 21 + 21 + 21 + UTF-8 + + + + + + org.springframework.boot + spring-boot-starter-web + + + + + org.springframework.boot + spring-boot-starter-validation + + + + + org.projectlombok + lombok + true + + + + + org.springframework.boot + spring-boot-starter-test + test + + + + + + + org.apache.maven.plugins + maven-compiler-plugin + 3.8.1 + + 21 + 21 + + + + org.springframework.boot + spring-boot-maven-plugin + + + + org.projectlombok + lombok + + + + + + + diff --git a/springboot-netspeed-limit/src/main/java/com/example/netspeed/BandwidthLimitApplication.java b/springboot-netspeed-limit/src/main/java/com/example/netspeed/BandwidthLimitApplication.java new file mode 100644 index 0000000..7ca7de7 --- /dev/null +++ b/springboot-netspeed-limit/src/main/java/com/example/netspeed/BandwidthLimitApplication.java @@ -0,0 +1,15 @@ +package com.example.netspeed; + +import org.springframework.boot.SpringApplication; +import org.springframework.boot.autoconfigure.SpringBootApplication; + +/** + * Spring Boot Bandwidth Limit Application + */ +@SpringBootApplication +public class BandwidthLimitApplication { + + public static void main(String[] args) { + SpringApplication.run(BandwidthLimitApplication.class, args); + } +} diff --git a/springboot-netspeed-limit/src/main/java/com/example/netspeed/annotation/BandwidthLimit.java b/springboot-netspeed-limit/src/main/java/com/example/netspeed/annotation/BandwidthLimit.java new file mode 100644 index 0000000..184c1d9 --- /dev/null +++ b/springboot-netspeed-limit/src/main/java/com/example/netspeed/annotation/BandwidthLimit.java @@ -0,0 +1,75 @@ +package com.example.netspeed.annotation; + +import com.example.netspeed.annotation.BandwidthUnit; +import com.example.netspeed.annotation.LimitType; + +import java.lang.annotation.ElementType; +import java.lang.annotation.Retention; +import java.lang.annotation.RetentionPolicy; +import java.lang.annotation.Target; + +/** + * 带宽限速注解 + * + * 支持多种限速维度： + * - GLOBAL: 全局限速，所有请求共享限速桶 + * - API: 按接口限速，每个接口独立限速 + * - USER: 按用户限速，根据用户标识（如请求头 X-User-Id）限速 + * - IP: 按IP限速，根据请求IP限速 + * + * 使用示例： + * 
+ * // 限速 200 KB/s
+ * {@code @BandwidthLimit(value = 200, unit = BandwidthUnit.KB)}
+ *
+ * // 按用户限速，免费用户 200KB/s，VIP用户 1MB/s
+ * {@code @BandwidthLimit(value = 200, unit = BandwidthUnit.KB, type = LimitType.USER)}
+ *
+ */ +@Target({ElementType.METHOD, ElementType.TYPE}) +@Retention(RetentionPolicy.RUNTIME) +public @interface BandwidthLimit { + + /** + * 限速值 + */ + long value() default 200; + + /** + * 限速单位 + */ + BandwidthUnit unit() default BandwidthUnit.KB; + + /** + * 限速类型 + */ + LimitType type() default LimitType.GLOBAL; + + /** + * 免费用户限速值（-1表示不区分） + */ + long free() default -1; + + /** + * VIP用户限速值（-1表示不区分） + */ + long vip() default -1; + + /** + * 桶容量倍数（相对于填充速率） + * 1.0 表示桶容量 = 1秒流量 + * 0.5 表示桶容量 = 0.5秒流量（更平滑） + * 2.0 表示桶容量 = 2秒流量（允许更大突发） + */ + double capacityMultiplier() default 1.0; + + /** + * 分块大小（字节），-1 表示自动计算 + */ + int chunkSize() default -1; + + /** + * 用户标识请求头名称（用于 USER 类型限速） + */ + String userHeader() default "X-User-Id"; +} diff --git a/springboot-netspeed-limit/src/main/java/com/example/netspeed/annotation/BandwidthUnit.java b/springboot-netspeed-limit/src/main/java/com/example/netspeed/annotation/BandwidthUnit.java new file mode 100644 index 0000000..094909f --- /dev/null +++ b/springboot-netspeed-limit/src/main/java/com/example/netspeed/annotation/BandwidthUnit.java @@ -0,0 +1,37 @@ +package com.example.netspeed.annotation; + +/** + * 带宽单位枚举 + */ +public enum BandwidthUnit { + B(1), + KB(1024), + MB(1024 * 1024), + GB(1024 * 1024 * 1024); + + private final long bytesPerSecond; + + BandwidthUnit(long bytesPerSecond) { + this.bytesPerSecond = bytesPerSecond; + } + + public long toBytesPerSecond(long value) { + return value * bytesPerSecond; + } + + public long getBytesPerUnit() { + return bytesPerSecond; + } + + public static String formatBytes(long bytes) { + if (bytes < KB.getBytesPerUnit()) { + return bytes + " B"; + } else if (bytes < MB.getBytesPerUnit()) { + return String.format("%.2f KB", bytes / (double) KB.getBytesPerUnit()); + } else if (bytes < GB.getBytesPerUnit()) { + return String.format("%.2f MB", bytes / (double) MB.getBytesPerUnit()); + } else { + return String.format("%.2f GB", bytes / (double) GB.getBytesPerUnit()); + } + } +} diff --git a/springboot-netspeed-limit/src/main/java/com/example/netspeed/annotation/LimitType.java b/springboot-netspeed-limit/src/main/java/com/example/netspeed/annotation/LimitType.java new file mode 100644 index 0000000..8b64014 --- /dev/null +++ b/springboot-netspeed-limit/src/main/java/com/example/netspeed/annotation/LimitType.java @@ -0,0 +1,23 @@ +package com.example.netspeed.annotation; + +/** + * 限速类型枚举 + */ +public enum LimitType { + /** + * 全局限速 - 所有请求共享限速桶 + */ + GLOBAL, + /** + * 按接口限速 - 每个接口独立限速 + */ + API, + /** + * 按用户限速 - 根据用户标识（如用户ID、token）限速 + */ + USER, + /** + * 按IP限速 - 根据请求IP限速 + */ + IP +} diff --git a/springboot-netspeed-limit/src/main/java/com/example/netspeed/config/BandwidthLimitConfig.java b/springboot-netspeed-limit/src/main/java/com/example/netspeed/config/BandwidthLimitConfig.java new file mode 100644 index 0000000..638ee7a --- /dev/null +++ b/springboot-netspeed-limit/src/main/java/com/example/netspeed/config/BandwidthLimitConfig.java @@ -0,0 +1,36 @@ +package com.example.netspeed.config; + +import com.example.netspeed.manager.BandwidthLimitManager; +import com.example.netspeed.web.BandwidthLimitInterceptor; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.web.servlet.config.annotation.InterceptorRegistry; +import org.springframework.web.servlet.config.annotation.WebMvcConfigurer; + +/** + * 带宽限速配置类 + */ +@Configuration +public class BandwidthLimitConfig implements WebMvcConfigurer { + + private BandwidthLimitInterceptor bandwidthLimitInterceptor; + + @Override + public void addInterceptors(InterceptorRegistry registry) { + registry.addInterceptor(bandwidthLimitInterceptor()) + .addPathPatterns("/api/**"); + } + + @Bean + public BandwidthLimitInterceptor bandwidthLimitInterceptor() { + if (bandwidthLimitInterceptor == null) { + bandwidthLimitInterceptor = new BandwidthLimitInterceptor(); + } + return bandwidthLimitInterceptor; + } + + @Bean + public BandwidthLimitManager bandwidthLimitManager() { + return new BandwidthLimitManager(); + } +} diff --git a/springboot-netspeed-limit/src/main/java/com/example/netspeed/controller/TestController.java b/springboot-netspeed-limit/src/main/java/com/example/netspeed/controller/TestController.java new file mode 100644 index 0000000..5518d60 --- /dev/null +++ b/springboot-netspeed-limit/src/main/java/com/example/netspeed/controller/TestController.java @@ -0,0 +1,217 @@ +package com.example.netspeed.controller; + +import com.example.netspeed.annotation.BandwidthLimit; +import com.example.netspeed.annotation.BandwidthUnit; +import com.example.netspeed.annotation.LimitType; +import com.example.netspeed.manager.BandwidthLimitManager; +import com.example.netspeed.web.BandwidthLimitHelper; +import com.example.netspeed.web.BandwidthLimitInterceptor; +import jakarta.servlet.http.HttpServletRequest; +import jakarta.servlet.http.HttpServletResponse; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.web.bind.annotation.*; + +import java.io.IOException; +import java.nio.charset.StandardCharsets; +import java.util.HashMap; +import java.util.Map; + +/** + * 测试控制器 - 提供各种限速维度的测试接口 + */ +@RestController +@RequestMapping("/api") +public class TestController { + + @Autowired(required = false) + private BandwidthLimitInterceptor bandwidthLimitInterceptor; + + /** + * 全局限速测试 - 200 KB/s + */ + @BandwidthLimit(value = 200, unit = BandwidthUnit.KB, type = LimitType.GLOBAL) + @GetMapping("/download/global") + public void downloadGlobal(HttpServletRequest request, HttpServletResponse response) throws IOException { + HttpServletResponse limitedResponse = BandwidthLimitHelper.getLimitedResponse(request, response); + limitedResponse.setContentType("application/octet-stream"); + limitedResponse.setHeader("Content-Disposition", "attachment; filename=global-limit-test.bin"); + limitedResponse.setHeader("X-Test-Type", "Global Limit"); + generateTestData(limitedResponse, 5 * 1024 * 1024); // 5MB + } + + /** + * API维度限速测试 - 500 KB/s + */ + @BandwidthLimit(value = 500, unit = BandwidthUnit.KB, type = LimitType.API) + @GetMapping("/download/api") + public void downloadByApi(HttpServletRequest request, HttpServletResponse response) throws IOException { + HttpServletResponse limitedResponse = BandwidthLimitHelper.getLimitedResponse(request, response); + limitedResponse.setContentType("application/octet-stream"); + limitedResponse.setHeader("Content-Disposition", "attachment; filename=api-limit-test.bin"); + limitedResponse.setHeader("X-Test-Type", "API Limit"); + generateTestData(limitedResponse, 5 * 1024 * 1024); // 5MB + } + + /** + * 用户维度限速测试 - 普通 200 KB/s，VIP 1 MB/s + */ + @BandwidthLimit(value = 200, unit = BandwidthUnit.KB, type = LimitType.USER, free = 200, vip = 1024) + @GetMapping("/download/user") + public void downloadByUser(HttpServletRequest request, + @RequestHeader(value = "X-User-Type", defaultValue = "free") String userType, + @RequestHeader(value = "X-User-Id", defaultValue = "anonymous") String userId, + HttpServletResponse response) throws IOException { + HttpServletResponse limitedResponse = BandwidthLimitHelper.getLimitedResponse(request, response); + limitedResponse.setContentType("application/octet-stream"); + limitedResponse.setHeader("Content-Disposition", "attachment; filename=user-limit-test.bin"); + limitedResponse.setHeader("X-Test-Type", "User Limit - " + userType); + limitedResponse.setHeader("X-User-Type", userType); + limitedResponse.setHeader("X-User-Id", userId); + generateTestData(limitedResponse, 5 * 1024 * 1024); // 5MB + } + + /** + * IP维度限速测试 - 300 KB/s + */ + @BandwidthLimit(value = 300, unit = BandwidthUnit.KB, type = LimitType.IP) + @GetMapping("/download/ip") + public void downloadByIp(HttpServletRequest request, HttpServletResponse response) throws IOException { + HttpServletResponse limitedResponse = BandwidthLimitHelper.getLimitedResponse(request, response); + limitedResponse.setContentType("application/octet-stream"); + limitedResponse.setHeader("Content-Disposition", "attachment; filename=ip-limit-test.bin"); + limitedResponse.setHeader("X-Test-Type", "IP Limit"); + generateTestData(limitedResponse, 5 * 1024 * 1024); // 5MB + } + + /** + * 自定义限速测试 + */ + @GetMapping("/download/custom") + public void downloadCustom(@RequestParam long bandwidth, + @RequestParam(defaultValue = "KB") String unit, + @RequestParam(defaultValue = "GLOBAL") String type, + HttpServletResponse response) throws IOException { + BandwidthUnit bandwidthUnit = BandwidthUnit.valueOf(unit.toUpperCase()); + LimitType limitType = LimitType.valueOf(type.toUpperCase()); + + response.setContentType("application/json"); + response.setHeader("X-Test-Type", "Custom Limit"); + response.setHeader("X-Bandwidth", bandwidth + " " + unit); + response.setHeader("X-Limit-Type", type); + + Map result = new HashMap<>(); + result.put("message", "Custom bandwidth limit request"); + result.put("bandwidth", bandwidth); + result.put("unit", unit); + result.put("type", type); + result.put("note", "This endpoint shows the parameters. Use the annotated endpoints for actual limiting."); + + response.getWriter().write(toJson(result)); + } + + /** + * 获取限速统计信息 + */ + @GetMapping("/stats") + public Map getStats() { + Map stats = new HashMap<>(); + + if (bandwidthLimitInterceptor != null) { + BandwidthLimitManager.BandwidthLimitStats limitStats = bandwidthLimitInterceptor.getStats(); + stats.put("globalCapacity", BandwidthUnit.formatBytes(limitStats.globalCapacity())); + stats.put("globalRefillRate", BandwidthUnit.formatBytes(limitStats.globalRefillRate()) + "/s"); + stats.put("globalAvailableTokens", BandwidthUnit.formatBytes(limitStats.globalAvailableTokens())); + stats.put("globalBytesConsumed", BandwidthUnit.formatBytes(limitStats.globalBytesConsumed())); + stats.put("globalActualRate", BandwidthUnit.formatBytes((long) limitStats.globalActualRate()) + "/s"); + stats.put("globalUtilization", String.format("%.1f%%", limitStats.globalUtilization() * 100)); + stats.put("apiBucketCount", limitStats.apiBucketCount()); + stats.put("userBucketCount", limitStats.userBucketCount()); + stats.put("ipBucketCount", limitStats.ipBucketCount()); + } else { + stats.put("error", "BandwidthLimitInterceptor not available"); + } + + return stats; + } + + /** + * 重置全局限速 + */ + @PostMapping("/reset/global") + public Map resetGlobal() { + Map result = new HashMap<>(); + if (bandwidthLimitInterceptor != null) { + bandwidthLimitInterceptor.resetGlobalBucket(); + result.put("status", "success"); + result.put("message", "Global bandwidth limit bucket reset"); + } else { + result.put("status", "error"); + result.put("message", "BandwidthLimitInterceptor not available"); + } + return result; + } + + /** + * 清除所有限速桶 + */ + @PostMapping("/reset/all") + public Map resetAll() { + Map result = new HashMap<>(); + if (bandwidthLimitInterceptor != null) { + bandwidthLimitInterceptor.clearAllBuckets(); + result.put("status", "success"); + result.put("message", "All bandwidth limit buckets cleared"); + } else { + result.put("status", "error"); + result.put("message", "BandwidthLimitInterceptor not available"); + } + return result; + } + + /** + * 生成测试数据 + */ + private void generateTestData(HttpServletResponse response, int size) throws IOException { + response.setContentLengthLong(size); + + byte[] buffer = new byte[8192]; + byte[] pattern = "This is a bandwidth limit test data. ".getBytes(StandardCharsets.UTF_8); + + int patternPos = 0; + int remaining = size; + + while (remaining > 0) { + int chunkSize = Math.min(buffer.length, remaining); + for (int i = 0; i < chunkSize; i++) { + buffer[i] = pattern[patternPos]; + patternPos = (patternPos + 1) % pattern.length; + } + response.getOutputStream().write(buffer, 0, chunkSize); + remaining -= chunkSize; + } + + response.getOutputStream().flush(); + } + + private String toJson(Map map) { + StringBuilder sb = new StringBuilder("{"); + boolean first = true; + for (Map.Entry entry : map.entrySet()) { + if (!first) { + sb.append(","); + } + sb.append("\"").append(entry.getKey()).append("\":"); + Object value = entry.getValue(); + if (value instanceof String) { + sb.append("\"").append(value).append("\""); + } else if (value instanceof Number) { + sb.append(value); + } else { + sb.append("\"").append(value).append("\""); + } + first = false; + } + sb.append("}"); + return sb.toString(); + } +} diff --git a/springboot-netspeed-limit/src/main/java/com/example/netspeed/core/RateLimitedOutputStream.java b/springboot-netspeed-limit/src/main/java/com/example/netspeed/core/RateLimitedOutputStream.java new file mode 100644 index 0000000..7766b3d --- /dev/null +++ b/springboot-netspeed-limit/src/main/java/com/example/netspeed/core/RateLimitedOutputStream.java @@ -0,0 +1,256 @@ +package com.example.netspeed.core; + +import jakarta.servlet.ServletOutputStream; +import jakarta.servlet.WriteListener; +import lombok.extern.slf4j.Slf4j; + +import java.io.IOException; +import java.io.OutputStream; + +/** + * 限速输出流（支持分块写入） + * + * 使用令牌桶算法控制写入速率，实现精确的带宽限速 + */ +@Slf4j +public class RateLimitedOutputStream extends ServletOutputStream { + + private final OutputStream outputStream; + private final TokenBucket tokenBucket; + private final int chunkSize; + private final long bandwidthBytesPerSecond; + + // 统计信息 + private long totalBytesWritten = 0; + private final long startTime = System.nanoTime(); + private volatile boolean closed = false; + private boolean logged = false; + + public RateLimitedOutputStream(OutputStream outputStream, long bandwidthBytesPerSecond) { + this(outputStream, bandwidthBytesPerSecond, calculateOptimalChunkSize(bandwidthBytesPerSecond)); + } + + /** + * 使用已有的 TokenBucket（共享限速状态） + * + * @param outputStream 底层输出流 + * @param tokenBucket 共享的令牌桶 + * @param bandwidthBytesPerSecond 限速（字节/秒） + */ + public RateLimitedOutputStream(OutputStream outputStream, + TokenBucket tokenBucket, + long bandwidthBytesPerSecond) { + this(outputStream, tokenBucket, bandwidthBytesPerSecond, calculateOptimalChunkSize(bandwidthBytesPerSecond)); + } + + /** + * 使用已有的 TokenBucket（共享限速状态），指定分块大小 + * + * @param outputStream 底层输出流 + * @param tokenBucket 共享的令牌桶 + * @param bandwidthBytesPerSecond 限速（字节/秒） + * @param chunkSize 分块大小 + */ + public RateLimitedOutputStream(OutputStream outputStream, + TokenBucket tokenBucket, + long bandwidthBytesPerSecond, + int chunkSize) { + this.outputStream = outputStream; + this.bandwidthBytesPerSecond = bandwidthBytesPerSecond; + this.chunkSize = Math.max(512, Math.min(chunkSize, 65536)); + this.tokenBucket = tokenBucket; + + log.info("RateLimitedOutputStream created with shared bucket: bandwidth={}/s, chunkSize={}", + formatBytes(bandwidthBytesPerSecond), chunkSize); + } + + /** + * @param outputStream 底层输出流 + * @param bandwidthBytesPerSecond 限速（字节/秒） + * @param chunkSize 分块大小，越小越平滑 + */ + public RateLimitedOutputStream(OutputStream outputStream, + long bandwidthBytesPerSecond, + int chunkSize) { + this.outputStream = outputStream; + this.bandwidthBytesPerSecond = bandwidthBytesPerSecond; + this.chunkSize = Math.max(512, Math.min(chunkSize, 65536)); + + // 桶容量 = 1秒流量，允许短时突发 + long capacity = bandwidthBytesPerSecond; + this.tokenBucket = new TokenBucket(capacity, bandwidthBytesPerSecond); + + log.info("RateLimitedOutputStream created: bandwidth={}/s, chunkSize={}, capacity={}/s", + formatBytes(bandwidthBytesPerSecond), chunkSize, formatBytes(capacity)); + } + + /** + * 计算最佳分块大小 + * 经验公式：chunkSize = bandwidthBytesPerSecond / 50 + */ + private static int calculateOptimalChunkSize(long bandwidthBytesPerSecond) { + if (bandwidthBytesPerSecond < 200 * 1024) { + // 低于 200KB/s，使用 1-4KB + return 1024; + } else if (bandwidthBytesPerSecond < 1024 * 1024) { + // 200KB/s - 1MB/s，使用 4-8KB + return 4096; + } else if (bandwidthBytesPerSecond < 5 * 1024 * 1024) { + // 1MB/s - 5MB/s，使用 8-16KB + return 8192; + } else { + // 高于 5MB/s，使用 16-32KB + return 16384; + } + } + + private String formatBytes(long bytes) { + if (bytes < 1024) { + return bytes + " B"; + } else if (bytes < 1024 * 1024) { + return String.format("%.2f KB", bytes / 1024.0); + } else if (bytes < 1024 * 1024 * 1024) { + return String.format("%.2f MB", bytes / (1024.0 * 1024)); + } else { + return String.format("%.2f GB", bytes / (1024.0 * 1024 * 1024)); + } + } + + @Override + public void write(int b) throws IOException { + checkClosed(); + tokenBucket.acquire(1); + outputStream.write(b); + totalBytesWritten++; + } + + @Override + public void write(byte[] b) throws IOException { + write(b, 0, b.length); + } + + @Override + public void write(byte[] b, int off, int len) throws IOException { + checkClosed(); + if (len == 0) { + return; + } + + if (!logged) { + log.info("RateLimitedOutputStream.write() called with len={} bytes", len); + logged = true; + } + + // 分块写入，使流量更平滑 + int remaining = len; + int offset = off; + + while (remaining > 0) { + int size = Math.min(chunkSize, remaining); + tokenBucket.acquire(size); + outputStream.write(b, offset, size); + offset += size; + remaining -= size; + totalBytesWritten += size; + } + + if (totalBytesWritten % (1024 * 1024) == 0) { + double elapsed = (System.nanoTime() - startTime) / 1_000_000_000.0; + double rate = elapsed > 0 ? (totalBytesWritten / elapsed) / 1024.0 : 0; + log.info("Written {} bytes, actual rate: {} KB/s", totalBytesWritten, String.format("%.2f", rate)); + } + } + + @Override + public void flush() throws IOException { + checkClosed(); + outputStream.flush(); + } + + @Override + public void close() throws IOException { + if (!closed) { + closed = true; + double elapsed = (System.nanoTime() - startTime) / 1_000_000_000.0; + double rate = elapsed > 0 ? (totalBytesWritten / elapsed) / 1024.0 : 0; + log.info("RateLimitedOutputStream closing: total bytes={}, elapsed={}s, rate={} KB/s", + totalBytesWritten, String.format("%.2f", elapsed), String.format("%.2f", rate)); + outputStream.flush(); + outputStream.close(); + } + } + + private void checkClosed() throws IOException { + if (closed) { + throw new IOException("Stream is closed"); + } + } + + @Override + public boolean isReady() { + return !closed; + } + + @Override + public void setWriteListener(WriteListener writeListener) { + throw new UnsupportedOperationException("Async write not supported"); + } + + /** + * 动态调整带宽 + */ + public void setBandwidth(long newBandwidth) { + tokenBucket.setRefillRate(newBandwidth); + } + + /** + * 获取当前可用令牌 + */ + public long getAvailableTokens() { + return tokenBucket.getAvailableTokens(); + } + + /** + * 获取实际传输速率 + */ + public double getActualRate() { + long elapsedNanos = System.nanoTime() - startTime; + if (elapsedNanos <= 0) { + return 0; + } + long elapsedSeconds = elapsedNanos / 1_000_000_000L; + return elapsedSeconds > 0 ? (double) totalBytesWritten / elapsedSeconds : 0; + } + + /** + * 获取总写入字节数 + */ + public long getTotalBytesWritten() { + return totalBytesWritten; + } + + /** + * 获取配置的带宽 + */ + public long getBandwidthBytesPerSecond() { + return bandwidthBytesPerSecond; + } + + /** + * 获取分块大小 + */ + public int getChunkSize() { + return chunkSize; + } + + /** + * 获取令牌桶利用率 + */ + public double getBucketUtilization() { + return tokenBucket.getUtilization(); + } + + public TokenBucket getTokenBucket() { + return tokenBucket; + } +} diff --git a/springboot-netspeed-limit/src/main/java/com/example/netspeed/core/TokenBucket.java b/springboot-netspeed-limit/src/main/java/com/example/netspeed/core/TokenBucket.java new file mode 100644 index 0000000..9351d56 --- /dev/null +++ b/springboot-netspeed-limit/src/main/java/com/example/netspeed/core/TokenBucket.java @@ -0,0 +1,189 @@ +package com.example.netspeed.core; + +import java.util.concurrent.locks.LockSupport; + +/** + * 令牌桶算法实现 + * + * 核心原理： + * 1. 桶容量：允许的突发流量上限 + * 2. 填充速率：长期平均传输速度 + * 3. 获取令牌：消耗对应数量的令牌，不足则等待 + */ +public class TokenBucket { + + private final long capacity; // 桶容量（字节） + private final long initialRefillRate; // 初始填充速率（字节/秒） + private volatile long refillRate; // 当前填充速率（字节/秒） + private volatile long tokens; // 当前令牌数（字节） + private volatile long lastRefillTime; // 上次填充时间（纳秒） + + // 统计信息 + private volatile long totalBytesConsumed; + private volatile long totalWaitTimeNanos; + private final long creationTime; + + public TokenBucket(long capacity, long refillRate) { + this.capacity = capacity; + this.initialRefillRate = refillRate; + this.refillRate = refillRate; + this.tokens = capacity; + this.lastRefillTime = System.nanoTime(); + this.totalBytesConsumed = 0; + this.totalWaitTimeNanos = 0; + this.creationTime = System.nanoTime(); + } + + /** + * 获取令牌（阻塞等待） + * + * @param permits 需要的令牌数（字节数） + */ + public synchronized void acquire(long permits) { + if (permits <= 0) { + return; + } + + long waitTime = refillAndCalculateWait(permits); + + if (waitTime > 0) { + sleepNanos(waitTime); + totalWaitTimeNanos += waitTime; + // 等待后再次填充并消费 + refill(); + tokens = Math.max(0, tokens - permits); + } else { + tokens -= permits; + } + + totalBytesConsumed += permits; + } + + /** + * 尝试获取令牌（非阻塞） + * + * @param permits 需要的令牌数 + * @return 是否成功获取 + */ + public synchronized boolean tryAcquire(long permits) { + if (permits <= 0) { + return true; + } + + refill(); + + if (tokens >= permits) { + tokens -= permits; + totalBytesConsumed += permits; + return true; + } + + return false; + } + + /** + * 填充令牌并计算需要等待的时间 + */ + private long refillAndCalculateWait(long permits) { + refill(); + + if (tokens >= permits) { + return 0; + } + + // 令牌不足，计算需要等待的时间 + long deficit = permits - tokens; + return (deficit * 1_000_000_000L) / refillRate; + } + + /** + * 填充令牌（核心逻辑） + */ + private void refill() { + long now = System.nanoTime(); + long elapsedNanos = now - lastRefillTime; + + if (elapsedNanos <= 0) { + return; + } + + // 根据时间差计算补充的令牌数 + long newTokens = (elapsedNanos * refillRate) / 1_000_000_000L; + + if (newTokens > 0) { + tokens = Math.min(capacity, tokens + newTokens); + lastRefillTime = now; + } + } + + /** + * 精确纳秒级等待 + */ + private void sleepNanos(long nanos) { + if (nanos <= 0) { + return; + } + + long end = System.nanoTime() + nanos; + while (System.nanoTime() < end) { + LockSupport.parkNanos(Math.max(1000, end - System.nanoTime())); + } + } + + /** + * 获取当前可用令牌数 + */ + public long getAvailableTokens() { + refill(); + return tokens; + } + + /** + * 动态调整填充速率 + */ + public synchronized void setRefillRate(long newRate) { + this.refillRate = newRate; + refill(); + } + + /** + * 重置令牌桶 + */ + public synchronized void reset() { + this.tokens = capacity; + this.refillRate = initialRefillRate; + this.lastRefillTime = System.nanoTime(); + } + + /** + * 获取实际传输速率 + */ + public double getActualRate() { + long elapsedNanos = System.nanoTime() - creationTime; + if (elapsedNanos <= 0) { + return 0; + } + long elapsedSeconds = elapsedNanos / 1_000_000_000L; + return elapsedSeconds > 0 ? (double) totalBytesConsumed / elapsedSeconds : 0; + } + + public long getCapacity() { + return capacity; + } + + public long getRefillRate() { + return refillRate; + } + + public long getTotalBytesConsumed() { + return totalBytesConsumed; + } + + public long getTotalWaitTimeNanos() { + return totalWaitTimeNanos; + } + + public double getUtilization() { + return capacity > 0 ? (double) tokens / capacity : 0; + } +} diff --git a/springboot-netspeed-limit/src/main/java/com/example/netspeed/manager/BandwidthLimitManager.java b/springboot-netspeed-limit/src/main/java/com/example/netspeed/manager/BandwidthLimitManager.java new file mode 100644 index 0000000..f7d89e8 --- /dev/null +++ b/springboot-netspeed-limit/src/main/java/com/example/netspeed/manager/BandwidthLimitManager.java @@ -0,0 +1,242 @@ +package com.example.netspeed.manager; + +import com.example.netspeed.annotation.LimitType; +import com.example.netspeed.core.TokenBucket; +import lombok.extern.slf4j.Slf4j; + +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.Executors; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.TimeUnit; + +/** + * 带宽限速管理器 + * + * 管理多维度的令牌桶： + * - GLOBAL: 全局共享一个令牌桶 + * - API: 每个接口路径一个令牌桶 + * - USER: 每个用户ID一个令牌桶 + * - IP: 每个IP地址一个令牌桶 + */ +@Slf4j +public class BandwidthLimitManager { + + // 全局限速桶 + private TokenBucket globalBucket; + + // API维度限速桶 (path -> TokenBucket) + private final ConcurrentHashMap apiBuckets = new ConcurrentHashMap<>(); + + // 用户维度限速桶 (userId -> TokenBucket) + private final ConcurrentHashMap userBuckets = new ConcurrentHashMap<>(); + + // IP维度限速桶 (ip -> TokenBucket) + private final ConcurrentHashMap ipBuckets = new ConcurrentHashMap<>(); + + // 定时清理服务 + private final ScheduledExecutorService cleanupExecutor = Executors.newSingleThreadScheduledExecutor(r -> { + Thread thread = new Thread(r, "bandwidth-limit-cleanup"); + thread.setDaemon(true); + return thread; + }); + + // 最后使用时间记录 + private final ConcurrentHashMap lastAccessTime = new ConcurrentHashMap<>(); + + // 空闲超时时间（毫秒） + private static final long IDLE_TIMEOUT_MS = 5 * 60 * 1000; // 5分钟 + + public BandwidthLimitManager() { + // 启动定时清理任务 + startCleanupTask(); + } + + /** + * 获取或创建令牌桶 + */ + public TokenBucket getBucket(LimitType type, String key, long capacity, long refillRate) { + return switch (type) { + case GLOBAL -> getGlobalBucket(capacity, refillRate); + case API -> getOrCreateBucket(apiBuckets, key, capacity, refillRate); + case USER -> getOrCreateBucket(userBuckets, key, capacity, refillRate); + case IP -> getOrCreateBucket(ipBuckets, key, capacity, refillRate); + }; + } + + /** + * 获取全局限速桶 + */ + private synchronized TokenBucket getGlobalBucket(long capacity, long refillRate) { + if (globalBucket == null) { + globalBucket = new TokenBucket(capacity, refillRate); + log.info("Created global bandwidth limit bucket: capacity={}, rate={}/s", + capacity, formatBytes(refillRate)); + } else if (globalBucket.getRefillRate() != refillRate) { + // 动态调整速率 + globalBucket.setRefillRate(refillRate); + log.info("Updated global bandwidth limit rate: {}/s", formatBytes(refillRate)); + } + return globalBucket; + } + + /** + * 获取或创建指定维度的令牌桶 + */ + private TokenBucket getOrCreateBucket(ConcurrentHashMap buckets, + String key, + long capacity, + long refillRate) { + return buckets.compute(key, (k, existing) -> { + if (existing == null) { + log.debug("Created new bandwidth limit bucket for {}: capacity={}, rate={}/s", + k, capacity, formatBytes(refillRate)); + return new TokenBucket(capacity, refillRate); + } + + // 更新最后访问时间 + lastAccessTime.put(key, System.currentTimeMillis()); + + // 动态调整速率 + if (existing.getRefillRate() != refillRate) { + existing.setRefillRate(refillRate); + log.debug("Updated bandwidth limit rate for {}: {}/s", k, formatBytes(refillRate)); + } + + return existing; + }); + } + + /** + * 启动定时清理任务 + */ + private void startCleanupTask() { + cleanupExecutor.scheduleAtFixedRate(() -> { + try { + cleanupIdleBuckets(); + } catch (Exception e) { + log.error("Error during cleanup", e); + } + }, 1, 1, TimeUnit.MINUTES); + } + + /** + * 清理空闲的令牌桶 + */ + private void cleanupIdleBuckets() { + long now = System.currentTimeMillis(); + + // 清理 API 维度 + cleanupMap(apiBuckets, now, "API"); + // 清理用户维度 + cleanupMap(userBuckets, now, "USER"); + // 清理 IP 维度 + cleanupMap(ipBuckets, now, "IP"); + + // 清理访问时间记录 + lastAccessTime.entrySet().removeIf(entry -> { + if (now - entry.getValue() > IDLE_TIMEOUT_MS) { + return true; + } + return false; + }); + } + + private void cleanupMap(ConcurrentHashMap buckets, long now, String type) { + buckets.keySet().removeIf(key -> { + Long lastAccess = lastAccessTime.get(key); + if (lastAccess == null || now - lastAccess > IDLE_TIMEOUT_MS) { + log.debug("Removed idle {} bandwidth bucket: {}", type, key); + lastAccessTime.remove(key); + return true; + } + return false; + }); + } + + /** + * 获取统计信息 + */ + public BandwidthLimitStats getStats() { + if (globalBucket != null) { + return new BandwidthLimitStats( + globalBucket.getCapacity(), + globalBucket.getRefillRate(), + globalBucket.getAvailableTokens(), + globalBucket.getTotalBytesConsumed(), + globalBucket.getActualRate(), + globalBucket.getTotalWaitTimeNanos(), + globalBucket.getUtilization(), + apiBuckets.size(), + userBuckets.size(), + ipBuckets.size() + ); + } + return new BandwidthLimitStats( + 0, 0, 0, 0, 0, 0, 0, 0, 0, 0 + ); + } + + /** + * 重置全局限速桶 + */ + public void resetGlobalBucket() { + if (globalBucket != null) { + globalBucket.reset(); + log.info("Reset global bandwidth limit bucket"); + } + } + + /** + * 清除所有维度的限速桶（除了全局） + */ + public void clearAllBuckets() { + apiBuckets.clear(); + userBuckets.clear(); + ipBuckets.clear(); + lastAccessTime.clear(); + log.info("Cleared all bandwidth limit buckets"); + } + + /** + * 关闭管理器 + */ + public void shutdown() { + cleanupExecutor.shutdown(); + try { + if (!cleanupExecutor.awaitTermination(5, TimeUnit.SECONDS)) { + cleanupExecutor.shutdownNow(); + } + } catch (InterruptedException e) { + cleanupExecutor.shutdownNow(); + Thread.currentThread().interrupt(); + } + } + + private String formatBytes(long bytes) { + if (bytes < 1024) { + return bytes + " B"; + } else if (bytes < 1024 * 1024) { + return String.format("%.2f KB", bytes / 1024.0); + } else if (bytes < 1024 * 1024 * 1024) { + return String.format("%.2f MB", bytes / (1024.0 * 1024)); + } else { + return String.format("%.2f GB", bytes / (1024.0 * 1024 * 1024)); + } + } + + /** + * 统计信息 + */ + public record BandwidthLimitStats( + long globalCapacity, // 全局桶容量 + long globalRefillRate, // 全局填充速率（字节/秒） + long globalAvailableTokens, // 全局可用令牌 + long globalBytesConsumed, // 全局已消耗字节 + double globalActualRate, // 全局实际传输速率（字节/秒） + long globalWaitTimeNanos, // 全局等待时间（纳秒） + double globalUtilization, // 全局利用率（0-1） + int apiBucketCount, // API限速桶数量 + int userBucketCount, // 用户限速桶数量 + int ipBucketCount // IP限速桶数量 + ) {} +} diff --git a/springboot-netspeed-limit/src/main/java/com/example/netspeed/web/BandwidthLimitHelper.java b/springboot-netspeed-limit/src/main/java/com/example/netspeed/web/BandwidthLimitHelper.java new file mode 100644 index 0000000..86de7d3 --- /dev/null +++ b/springboot-netspeed-limit/src/main/java/com/example/netspeed/web/BandwidthLimitHelper.java @@ -0,0 +1,35 @@ +package com.example.netspeed.web; + +import jakarta.servlet.http.HttpServletRequest; +import jakarta.servlet.http.HttpServletResponse; + +/** + * 带宽限速辅助类 + * + * 用于从请求中获取限速响应包装器 + */ +public class BandwidthLimitHelper { + + private static final String WRAPPED_RESPONSE_ATTR = "BandwidthLimitWrappedResponse"; + + /** + * 获取限速响应包装器（如果存在） + */ + public static HttpServletResponse getLimitedResponse(HttpServletRequest request, HttpServletResponse defaultResponse) { + BandwidthLimitResponseWrapper wrappedResponse = + (BandwidthLimitResponseWrapper) request.getAttribute(WRAPPED_RESPONSE_ATTR); + + if (wrappedResponse != null) { + return wrappedResponse; + } + + return defaultResponse; + } + + /** + * 检查是否应用了限速 + */ + public static boolean isLimited(HttpServletRequest request) { + return request.getAttribute(WRAPPED_RESPONSE_ATTR) != null; + } +} diff --git a/springboot-netspeed-limit/src/main/java/com/example/netspeed/web/BandwidthLimitInterceptor.java b/springboot-netspeed-limit/src/main/java/com/example/netspeed/web/BandwidthLimitInterceptor.java new file mode 100644 index 0000000..c060da3 --- /dev/null +++ b/springboot-netspeed-limit/src/main/java/com/example/netspeed/web/BandwidthLimitInterceptor.java @@ -0,0 +1,152 @@ +package com.example.netspeed.web; + +import com.example.netspeed.annotation.BandwidthLimit; +import com.example.netspeed.annotation.BandwidthUnit; +import com.example.netspeed.annotation.LimitType; +import com.example.netspeed.core.TokenBucket; +import com.example.netspeed.manager.BandwidthLimitManager; +import jakarta.servlet.http.HttpServletRequest; +import jakarta.servlet.http.HttpServletResponse; +import lombok.extern.slf4j.Slf4j; +import org.springframework.core.annotation.AnnotationUtils; +import org.springframework.web.method.HandlerMethod; +import org.springframework.web.servlet.HandlerInterceptor; + +/** + * 带宽限速拦截器 + * + * 在 preHandle 中包装响应，在 afterCompletion 中关闭 + */ +@Slf4j +public class BandwidthLimitInterceptor implements HandlerInterceptor { + + private final BandwidthLimitManager limitManager = new BandwidthLimitManager(); + + private static final String WRAPPED_RESPONSE_ATTR = "BandwidthLimitWrappedResponse"; + private static final String ORIGINAL_RESPONSE_ATTR = "BandwidthLimitOriginalResponse"; + + @Override + public boolean preHandle(HttpServletRequest request, HttpServletResponse response, Object handler) + throws Exception { + + if (!(handler instanceof HandlerMethod)) { + return true; + } + + HandlerMethod handlerMethod = (HandlerMethod) handler; + BandwidthLimit annotation = handlerMethod.getMethodAnnotation(BandwidthLimit.class); + + // 如果方法没有注解，检查类级别的注解 + if (annotation == null) { + annotation = AnnotationUtils.findAnnotation(handlerMethod.getBeanType(), BandwidthLimit.class); + } + + if (annotation != null) { + String path = request.getRequestURI(); + log.info("========== Interceptor: Found @BandwidthLimit for path: {}, type: {}, value: {} {}/s ==========", + path, annotation.type(), annotation.value(), annotation.unit()); + + // 获取带宽参数 + LimitType type = annotation.type(); + long bandwidth = calculateBandwidth(request, annotation); + long bandwidthBytesPerSecond = annotation.unit().toBytesPerSecond(bandwidth); + long capacity = (long) (bandwidthBytesPerSecond * annotation.capacityMultiplier()); + String key = getLimitKey(request, type, path, annotation); + + // 获取或创建令牌桶 + TokenBucket bucket = limitManager.getBucket(type, key, capacity, bandwidthBytesPerSecond); + + log.info("Interceptor: Token bucket created - type={}, key={}, capacity={}/s, rate={}/s", + type, key, BandwidthUnit.formatBytes(capacity), BandwidthUnit.formatBytes(bandwidthBytesPerSecond)); + + // 设置响应头到原始响应（这样浏览器才能看到） + response.setHeader("X-Bandwidth-Limit", BandwidthUnit.formatBytes(bandwidthBytesPerSecond) + "/s"); + response.setHeader("X-Bandwidth-Type", type.name()); + response.setHeader("X-Bandwidth-Key", key); + response.setHeader("X-Bandwidth-Capacity", BandwidthUnit.formatBytes(capacity)); + + log.info("Interceptor: Response headers set - X-Bandwidth-Limit={}", + BandwidthUnit.formatBytes(bandwidthBytesPerSecond) + "/s"); + + // 创建限速响应包装器（传入共享的 TokenBucket） + BandwidthLimitResponseWrapper wrappedResponse = new BandwidthLimitResponseWrapper( + response, bucket, bandwidthBytesPerSecond, annotation.chunkSize()); + + // 将包装器保存到请求中 + request.setAttribute(WRAPPED_RESPONSE_ATTR, wrappedResponse); + request.setAttribute(ORIGINAL_RESPONSE_ATTR, response); + request.setAttribute("BandwidthLimit", annotation); + } + + return true; + } + + @Override + public void afterCompletion(HttpServletRequest request, HttpServletResponse response, + Object handler, Exception ex) { + // 清理资源 + BandwidthLimitResponseWrapper wrappedResponse = + (BandwidthLimitResponseWrapper) request.getAttribute(WRAPPED_RESPONSE_ATTR); + if (wrappedResponse != null) { + try { + wrappedResponse.close(); + } catch (Exception e) { + log.error("Error closing wrapped response", e); + } + } + } + + private long calculateBandwidth(HttpServletRequest request, BandwidthLimit annotation) { + if (annotation.free() > 0 || annotation.vip() > 0) { + String userType = request.getHeader("X-User-Type"); + if ("vip".equalsIgnoreCase(userType)) { + return annotation.vip() > 0 ? annotation.vip() : annotation.value(); + } else if ("free".equalsIgnoreCase(userType)) { + return annotation.free() > 0 ? annotation.free() : annotation.value(); + } + } + return annotation.value(); + } + + private String getLimitKey(HttpServletRequest request, LimitType type, String path, BandwidthLimit annotation) { + return switch (type) { + case GLOBAL -> "global"; + case API -> path; + case USER -> { + String userId = request.getHeader(annotation.userHeader()); + yield userId != null ? userId : request.getRemoteAddr(); + } + case IP -> getClientIp(request); + }; + } + + private String getClientIp(HttpServletRequest request) { + String ip = request.getHeader("X-Forwarded-For"); + if (ip == null || ip.isEmpty() || "unknown".equalsIgnoreCase(ip)) { + ip = request.getHeader("X-Real-IP"); + } + if (ip == null || ip.isEmpty() || "unknown".equalsIgnoreCase(ip)) { + ip = request.getRemoteAddr(); + } + if (ip != null && ip.contains(",")) { + ip = ip.split(",")[0].trim(); + } + return ip; + } + + public BandwidthLimitManager.BandwidthLimitStats getStats() { + return limitManager.getStats(); + } + + public void resetGlobalBucket() { + limitManager.resetGlobalBucket(); + } + + public void clearAllBuckets() { + limitManager.clearAllBuckets(); + } + + public void shutdown() { + limitManager.shutdown(); + } +} diff --git a/springboot-netspeed-limit/src/main/java/com/example/netspeed/web/BandwidthLimitResponseWrapper.java b/springboot-netspeed-limit/src/main/java/com/example/netspeed/web/BandwidthLimitResponseWrapper.java new file mode 100644 index 0000000..0be45f2 --- /dev/null +++ b/springboot-netspeed-limit/src/main/java/com/example/netspeed/web/BandwidthLimitResponseWrapper.java @@ -0,0 +1,160 @@ +package com.example.netspeed.web; + +import com.example.netspeed.core.RateLimitedOutputStream; +import com.example.netspeed.core.TokenBucket; +import jakarta.servlet.ServletOutputStream; +import jakarta.servlet.http.HttpServletResponse; +import jakarta.servlet.http.HttpServletResponseWrapper; +import lombok.extern.slf4j.Slf4j; + +import java.io.IOException; +import java.io.OutputStreamWriter; +import java.io.PrintWriter; + +/** + * 带宽限速响应包装器 + * + * 包装 HttpServletResponse 的 OutputStream，使用 RateLimitedOutputStream 实现限速 + */ +@Slf4j +public class BandwidthLimitResponseWrapper extends HttpServletResponseWrapper { + + private final long bandwidthBytesPerSecond; + private final int chunkSize; + private final TokenBucket sharedTokenBucket; + private RateLimitedOutputStream limitedOutputStream; + private PrintWriter writer; + private boolean outputStreamUsed = false; + private boolean headersCopied = false; + + public BandwidthLimitResponseWrapper(HttpServletResponse response, long bandwidthBytesPerSecond) { + this(response, null, bandwidthBytesPerSecond, -1); + } + + public BandwidthLimitResponseWrapper(HttpServletResponse response, long bandwidthBytesPerSecond, int chunkSize) { + this(response, null, bandwidthBytesPerSecond, chunkSize); + } + + public BandwidthLimitResponseWrapper(HttpServletResponse response, + TokenBucket tokenBucket, + long bandwidthBytesPerSecond, + int chunkSize) { + super(response); + this.sharedTokenBucket = tokenBucket; + this.bandwidthBytesPerSecond = bandwidthBytesPerSecond; + this.chunkSize = chunkSize; + } + + private String formatBytes(long bytes) { + if (bytes < 1024) { + return bytes + " B"; + } else if (bytes < 1024 * 1024) { + return String.format("%.2f KB", bytes / 1024.0); + } else if (bytes < 1024 * 1024 * 1024) { + return String.format("%.2f MB", bytes / (1024.0 * 1024)); + } else { + return String.format("%.2f GB", bytes / (1024.0 * 1024 * 1024)); + } + } + + @Override + public ServletOutputStream getOutputStream() throws IOException { + if (!outputStreamUsed) { + log.info("BandwidthLimitResponseWrapper.getOutputStream() called, bandwidth={}/s, sharedBucket={}", + formatBytes(bandwidthBytesPerSecond), sharedTokenBucket != null); + outputStreamUsed = true; + } + if (limitedOutputStream == null) { + if (sharedTokenBucket != null) { + // 使用共享的 TokenBucket + if (chunkSize > 0) { + limitedOutputStream = new RateLimitedOutputStream( + super.getOutputStream(), + sharedTokenBucket, + bandwidthBytesPerSecond, + chunkSize + ); + } else { + limitedOutputStream = new RateLimitedOutputStream( + super.getOutputStream(), + sharedTokenBucket, + bandwidthBytesPerSecond + ); + } + } else { + // 创建新的 TokenBucket（兼容旧代码） + if (chunkSize > 0) { + limitedOutputStream = new RateLimitedOutputStream( + super.getOutputStream(), + bandwidthBytesPerSecond, + chunkSize + ); + } else { + limitedOutputStream = new RateLimitedOutputStream( + super.getOutputStream(), + bandwidthBytesPerSecond + ); + } + } + } + return limitedOutputStream; + } + + @Override + public PrintWriter getWriter() throws IOException { + if (writer == null) { + writer = new PrintWriter(new OutputStreamWriter(getOutputStream(), getCharacterEncoding()), true); + } + return writer; + } + + @Override + public void flushBuffer() throws IOException { + if (writer != null) { + writer.flush(); + } else if (limitedOutputStream != null) { + limitedOutputStream.flush(); + } + super.flushBuffer(); + } + + @Override + public void setContentType(String type) { + super.setContentType(type); + } + + @Override + public void setCharacterEncoding(String charset) { + super.setCharacterEncoding(charset); + } + + @Override + public void setHeader(String name, String value) { + super.setHeader(name, value); + } + + @Override + public void addHeader(String name, String value) { + super.addHeader(name, value); + } + + @Override + public void setIntHeader(String name, int value) { + super.setIntHeader(name, value); + } + + /** + * 获取限速输出流（用于获取统计信息） + */ + public RateLimitedOutputStream getRateLimitedOutputStream() { + return limitedOutputStream; + } + + public void close() throws IOException { + if (limitedOutputStream != null) { + log.info("BandwidthLimitResponseWrapper closing, total bytes: {}", + limitedOutputStream.getTotalBytesWritten()); + limitedOutputStream.close(); + } + } +} diff --git a/springboot-netspeed-limit/src/main/resources/application.yml b/springboot-netspeed-limit/src/main/resources/application.yml new file mode 100644 index 0000000..148325c --- /dev/null +++ b/springboot-netspeed-limit/src/main/resources/application.yml @@ -0,0 +1,11 @@ +server: + port: 8080 + +spring: + application: + name: bandwidth-limit + +logging: + level: + com.example.netspeed: DEBUG + org.springframework.web: INFO diff --git a/springboot-netspeed-limit/src/main/resources/static/index.html b/springboot-netspeed-limit/src/main/resources/static/index.html new file mode 100644 index 0000000..0f3d22e --- /dev/null +++ b/springboot-netspeed-limit/src/main/resources/static/index.html @@ -0,0 +1,419 @@ + + + + + + Spring Boot 带宽限速测试 + + + + +
+ +
+

+ Spring Boot 网络带宽限速 +

+

基于令牌桶算法的多维度流量控制

+
+ + +
+

+ + + + 限速统计信息 + - +

+
+
+
-
+
已传输字节
+
+
+
-
+
实际传输速率
+
+
+
-
+
令牌利用率
+
+
+
-
+
可用配额 (字节)
+
+
+
-
+
API限速桶
+
+
+
-
+
用户限速桶
+
+
+
+ + +
+ +
+
+

全局限速

+ 200 KB/s +
+

所有请求共享同一个限速桶，适合保护服务器整体带宽

+ + +
+ + +
+
+

API维度限速

+ 500 KB/s +
+

每个接口独立限速，不同接口的限速桶互不影响

+ + +
+ + +
+
+

用户维度限速

+ 200KB/s - 1MB/s +
+

根据用户类型限速，免费用户200KB/s，VIP用户1MB/s

+
+ + +
+ + +
+ + +
+
+

IP维度限速

+ 300 KB/s +
+

根据客户端IP限速，每个IP地址拥有独立的限速桶

+ + +
+
+ + +
+

+ + + + + 控制面板 +

+
+ + + +
+
+ + +
+

关于限速算法

+
+
+

令牌桶算法原理

+
    +
  • • 桶容量：允许的突发流量上限
    • +
  • • 填充速率：长期平均传输速度
    • +
  • • 分块大小：影响流量平滑度
    • +
+
+
+

应用场景

+
    +
  • • 文件下载服务的速度控制
    • +
  • • 视频流媒体的带宽管理
    • +
  • • API接口的响应限速
    • +
+
+
+
+
+ + + +