当前位置：首页 > article >正文

使用 Redisson 实现分布式并发限流

article 2025/2/6 4:49:26

使用 Redisson 实现简单的分布式并发限流

在开发高并发的分布式应用时，控制同时处理的任务数量至关重要。不同于限制每秒请求数（QPS），我们有时需要限制系统的并发任务数，以避免系统资源过度占用。Redisson 提供的 RPermitExpirableSemaphore 是一个理想的解决方案，它可以帮助我们实现分布式并发限流，并且具备自动回收超时许可和程序崩溃后释放资源的能力。

为什么选择 Redisson 的 RPermitExpirableSemaphore？

分布式控制：支持跨多个服务实例共享同一个信号量，确保所有实例都能公平竞争获取许可。
自动回收：为每个许可设置过期时间，超时后自动释放，防止死锁。
资源安全：即使服务崩溃，未释放的许可也能被自动清理，保证系统稳定运行。

快速入门

添加依赖

首先，在你的项目中添加 Redisson 的 Maven 依赖：

<!-- Maven -->
<dependency>
    <groupId>org.redisson</groupId>
    <artifactId>redisson</artifactId>
    <version>3.21.1</version>
</dependency>

基本用法

以下是一个简单的例子，展示了如何使用 RPermitExpirableSemaphore 来限制并发任务的数量：

import org.redisson.Redisson;
import org.redisson.api.RPermitExpirableSemaphore;
import org.redisson.api.RedissonClient;
import org.redisson.config.Config;

public class ConcurrentTaskLimiter {

    private final RedissonClient redissonClient;

    public ConcurrentTaskLimiter(String redisAddress) {
        Config config = new Config();
        config.useSingleServer().setAddress(redisAddress);
        this.redissonClient = Redisson.create(config);
    }

    public void executeWithLimit(Runnable task, int maxConcurrentTasks, long leaseTime, TimeUnit unit) throws InterruptedException {
        RPermitExpirableSemaphore semaphore = redissonClient.getPermitExpirableSemaphore("mySemaphore");
        semaphore.trySetPermits(maxConcurrentTasks); // 设置最大并发任务数
        
        String permitId = semaphore.acquire(leaseTime, unit); // 尝试获取许可
        if (permitId != null) {
            try {
                task.run(); // 执行任务
            } finally {
                semaphore.release(permitId); // 确保许可被释放
            }
        } else {
            System.out.println("Failed to acquire permit within timeout.");
        }
    }

    public static void main(String[] args) throws InterruptedException {
        ConcurrentTaskLimiter limiter = new ConcurrentTaskLimiter("redis://127.0.0.1:6379");
        
        Runnable task = () -> {
            System.out.println("Executing task...");
            try {
                Thread.sleep(5000); // 模拟耗时操作
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        };

        // 允许最多3个并发任务，每个许可的有效期为10秒
        limiter.executeWithLimit(task, 3, 10, TimeUnit.SECONDS);
    }
}