- ? 使用 Redis 記錄 所有用戶的實時并發下載數
- ? 使用 Bucket4j 實現 全局下載速率限制(動態)
- ? 支持 動態調整限速策略
- ? 下載接口安全、穩定、可監控
🧩 整體架構概覽
模塊 | 功能 |
---|---|
Redis | 存儲全局并發數和帶寬令牌桶狀態 |
Bucket4j + Redis | 分布式限速器(基于令牌桶算法) |
Spring Boot Web | 提供文件下載接口 |
AOP / Interceptor(可選) | 用于統一處理限流邏輯 |
📦 1. Maven 依賴(pom.xml
)
<dependencies><!-- Spring Boot --><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-web</artifactId></dependency><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-data-redis</artifactId></dependency><!-- Redis 連接池 --><dependency><groupId>io.lettuce.core</groupId><artifactId>lettuce-core</artifactId></dependency><!-- Bucket4j 核心與 Redis 集成 --><dependency><groupId>com.github.vladimir-bukhtoyarov</groupId><artifactId>bucket4j-core</artifactId><version>5.3.0</version></dependency><dependency><groupId>com.github.vladimir-bukhtoyarov</groupId><artifactId>bucket4j-redis</artifactId><version>5.3.0</version></dependency></dependencies>
🛠? 2. Redis 工具類:記錄全局并發數
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.core.script.DefaultRedisScript;
import org.springframework.stereotype.Component;import java.util.Collections;
import java.util.concurrent.TimeUnit;@Component
public class GlobalDownloadCounter {private final StringRedisTemplate redisTemplate;private final DefaultRedisScript<Long> incrScript;private final DefaultRedisScript<Long> decrScript;public static final String KEY_CONCURRENT = "global:download:concurrent";private static final long TTL_SECONDS = 60; // 自動清理僵尸計數public GlobalDownloadCounter(StringRedisTemplate redisTemplate) {this.redisTemplate = redisTemplate;// Lua 腳本:原子增加并發數并設置過期時間String scriptIncr = """local key = KEYS[1]local ttl = tonumber(ARGV[1])local count = redis.call('GET', key)if not count thenredis.call('SET', key, 1)redis.call('EXPIRE', key, ttl)return 1elsecount = tonumber(count) + 1redis.call('SET', key, count)redis.call('EXPIRE', key, ttl)return countend""";incrScript = new DefaultRedisScript<>(scriptIncr, Long.class);// Lua 腳本:原子減少并發數String scriptDecr = """local key = KEYS[1]local count = redis.call('GET', key)if not count or tonumber(count) <= 0 thenreturn 0elsecount = tonumber(count) - 1redis.call('SET', key, count)return countend""";decrScript = new DefaultRedisScript<>(scriptDecr, Long.class);}public long increment() {return redisTemplate.execute(incrScript, Collections.singletonList(KEY_CONCURRENT), TTL_SECONDS).longValue();}public long decrement() {return redisTemplate.execute(decrScript, Collections.singletonList(KEY_CONCURRENT)).longValue();}public long getCurrentCount() {String value = redisTemplate.opsForValue().get(KEY_CONCURRENT);return value == null ? 0 : Long.parseLong(value);}
}
?? 3. Bucket4j 配置:分布式限速器(帶 Redis)
import io.github.bucket4j.Bandwidth;
import io.github.bucket4j.Refill;
import io.github.bucket4j.distributed.proxy.ProxyManager;
import io.github.bucket4j.distributed.proxy.RedisProxyManager;
import io.github.bucket4j.redis.lettuce.cas.LettuceReactiveProxyManager;
import io.lettuce.core.RedisClient;
import io.lettuce.core.api.StatefulRedisConnection;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;import java.time.Duration;@Configuration
public class BandwidthLimiterConfig {@Beanpublic RedisClient redisClient() {return RedisClient.create("redis://localhost:6379");}@Beanpublic StatefulRedisConnection<String, String> redisConnection(RedisClient redisClient) {return redisClient.connect();}@Beanpublic ProxyManager<String> proxyManager(StatefulRedisConnection<String, String> connection) {return LettuceReactiveProxyManager.builder().build(connection.reactive());}@Beanpublic Bandwidth globalBandwidthLimit() {// 默認 10MB/sreturn Bandwidth.classic(10 * 1024 * 1024, Refill.greedy(10 * 1024 * 1024, Duration.ofSeconds(1)));}@Beanpublic Bucket globalBandwidthLimiter(ProxyManager<String> proxyManager, Bandwidth bandwidthLimit) {return proxyManager.builder().build("global:bandwidth:limiter", bandwidthLimit);}
}
📡 4. 下載接口實現
import io.github.bucket4j.Bucket;
import org.springframework.web.bind.annotation.*;import javax.servlet.http.HttpServletResponse;
import java.io.*;
import java.nio.file.Files;
import java.nio.file.Paths;@RestController
@RequestMapping("/api/download")
public class DownloadController {private static final int MAX_CONCURRENT_DOWNLOADS = 100;private final GlobalDownloadCounter downloadCounter;private final Bucket bandwidthLimiter;public DownloadController(GlobalDownloadCounter downloadCounter, Bucket bandwidthLimiter) {this.downloadCounter = downloadCounter;this.bandwidthLimiter = bandwidthLimiter;}@GetMapping("/{fileId}")public void downloadFile(@PathVariable String fileId, HttpServletResponse response) throws IOException {long currentCount = downloadCounter.getCurrentCount();if (currentCount >= MAX_CONCURRENT_DOWNLOADS) {response.setStatus(HttpServletResponse.SC_TOO_MANY_REQUESTS);response.getWriter().write("Too many downloads. Please try again later.");return;}downloadCounter.increment();try {// 設置響應頭response.setContentType("application/octet-stream");response.setHeader("Content-Disposition", "attachment; filename=" + fileId + ".bin");ServletOutputStream out = response.getOutputStream();// 文件路徑(示例)String filePath = "/path/to/files/" + fileId + ".bin";if (!Files.exists(Paths.get(filePath))) {response.setStatus(HttpServletResponse.SC_NOT_FOUND);response.getWriter().write("File not found.");return;}byte[] buffer = new byte[8192]; // 每次讀取 8KBRandomAccessFile file = new RandomAccessFile(filePath, "r");int bytesRead;while ((bytesRead = file.read(buffer)) != -1) {if (bytesRead > 0) {boolean consumed = bandwidthLimiter.tryConsume(bytesRead);if (!consumed) {Thread.sleep(100); // 等待令牌生成continue;}out.write(buffer, 0, bytesRead);out.flush();}}file.close();out.close();} catch (InterruptedException e) {Thread.currentThread().interrupt(); // 恢復中斷狀態} finally {downloadCounter.decrement();}}
}
🔁 5. 動態調整下載速率接口(可選)
@RestController
@RequestMapping("/api/limit")
public class RateLimitController {private final Bucket bandwidthLimiter;private final Bandwidth globalBandwidthLimit;public RateLimitController(Bucket bandwidthLimiter, Bandwidth globalBandwidthLimit) {this.bandwidthLimiter = bandwidthLimiter;this.globalBandwidthLimit = globalBandwidthLimit;}@PostMapping("/set-bandwidth")public String setBandwidth(@RequestParam int mbPerSecond) {Bandwidth newLimit = Bandwidth.classic(mbPerSecond * 1024 * 1024,Refill.greedy(mbPerSecond * 1024 * 1024, Duration.ofSeconds(1)));bandwidthLimiter.replaceConfiguration(newLimit);return "Global bandwidth limit updated to " + mbPerSecond + " MB/s";}
}
📊 6. 監控接口(可選)
@GetMapping("/monitor/concurrent")
public ResponseEntity<Long> getConcurrentDownloads() {return ResponseEntity.ok(downloadCounter.getCurrentCount());
}
🧪 7. 測試建議
你可以使用 curl
或 Postman 發起多并發請求測試:
for i in {1..200}; docurl -X GET "http://localhost:8080/api/download/file1" --output "file$i.bin" &
done
觀察是否觸發限流、并發控制是否生效。
? 總結
組件 | 作用 |
---|---|
Redis | 分布式存儲并發數和限流令牌桶 |
Lua 腳本 | 原子操作并發計數器 |
Bucket4j + Redis | 全局下載速率限制 |
Spring Boot Controller | 處理下載邏輯 |
try-finally | 保證資源釋放 |
動態接口 /set-bandwidth | 支持運行時修改限速 |
📌 擴展建議(可選)
- 將限流邏輯封裝到 AOP 切面 中
- 添加 Prometheus 指標暴露并發數、限流次數等
- 使用 Nginx 或 Gateway 做額外的限流保護
- 加入用戶身份識別,支持 用戶級限速
- 使用 Kafka 異步記錄日志或審計下載行為