slowbyte
首页友链关于

Redisson 限流与高级特性

发表于 2026-08-07 1263 字 7 min read

文章目录
分布式锁
Redisson 限流与高级特性 一、RRateLimiter（分布式限流器） Redisson 的限流器基于令牌桶算法，支持分布式场景下的平滑限流。 基本用法 RateType 说明 典型场景

Redisson 限流与高级特性

一、RRateLimiter（分布式限流器）

Redisson 的限流器基于令牌桶算法，支持分布式场景下的平滑限流。

基本用法

RRateLimiter limiter = redisson.getRateLimiter("rateLimiter:order");

// 初始化：每秒 10 个令牌（速率 = 10/s）
limiter.trySetRate(RateType.OVERALL, 10, 1, RateIntervalUnit.SECONDS);

// 获取令牌（阻塞直到有令牌）
limiter.acquire();

// 尝试获取令牌（非阻塞）
boolean acquired = limiter.tryAcquire();
boolean acquired = limiter.tryAcquire(5);  // 尝试获取 5 个令牌
boolean acquired = limiter.tryAcquire(1, 10, TimeUnit.SECONDS);  // 最多等 10 秒

// 获取等待时间
long waitTime = limiter.availablePermits();

RateType 说明

// 跨所有实例共享限流（全局限流）
RateType.OVERALL
// 例如：整个系统的下单接口限流 100/s

// 按客户端实例单独限流
RateType.PER_CLIENT
// 例如：每个 API 网关实例各自限流 100/s

典型场景

@Service
public class OrderService {

    @Autowired
    private RedissonClient redisson;

    @PostConstruct
    public void init() {
        RRateLimiter limiter = redisson.getRateLimiter("rate:createOrder");
        limiter.trySetRate(RateType.OVERALL, 100, 1, RateIntervalUnit.SECONDS);
    }

    public Result createOrder(OrderDTO dto) {
        RRateLimiter limiter = redisson.getRateLimiter("rate:createOrder");

        // 等待获取令牌（被限流时友好等待）
        limiter.acquire();

        // 实际业务
        return orderService.createOrder(dto);
    }
}

速率设置速查

// 每秒 100 个
limiter.trySetRate(RateType.OVERALL, 100, 1, RateIntervalUnit.SECONDS);

// 每分钟 1000 个
limiter.trySetRate(RateType.OVERALL, 1000, 1, RateIntervalUnit.MINUTES);

// 每小时 5000 个
limiter.trySetRate(RateType.OVERALL, 5000, 1, RateIntervalUnit.HOURS);

二、RPermitExpirableSemaphore vs RRateLimiter

特性RRateLimiterRPermitExpirableSemaphore
算法令牌桶（平滑限流）信号量（固定并发数）
突发流量✅ 允许一定突发❌ 严格按许可数
等待支持支持
适用场景API 限流、流量整形数据库连接池、并发任务数控制
// RRateLimiter：平滑限流
// 配置 10/s，瞬间 100 请求持续约 10s 处理完
RRateLimiter rl = redisson.getRateLimiter("api:rate");

// RSemaphore：控制最大并发
// 最多 10 个同时在执行
RSemaphore sp = redisson.getSemaphore("pool:concurrent");

三、RMapCache（带缓存的 Map）

RMapCache<String, Object> cache = redisson.getMapCache("myCache");

// 每个字段有自己的 TTL
cache.put("key1", "value1", 10, TimeUnit.MINUTES);
cache.put("key2", "value2", 1, TimeUnit.HOURS);

// 最大空闲时间
cache.put("key3", "value3", 10, TimeUnit.MINUTES, 5, TimeUnit.MINUTES);
// TTL: 10 分钟，空闲 5 分钟未访问则自动清除

// 基于比较的写入
cache.putIfAbsent("key4", "value4", 30, TimeUnit.SECONDS);

RLocalCachedMap（两级缓存）

LocalCachedMapOptions options = LocalCachedMapOptions.defaults()
    .evictionPolicy(LocalCachedMapOptions.EvictionPolicy.LFU)     // LFU 淘汰策略
    .cacheSize(1000)                                                // 本地缓存最多 1000 个
    .timeToLive(60, TimeUnit.SECONDS)                               // 本地缓存 TTL
    .syncStrategy(LocalCachedMapOptions.SyncStrategy.INVALIDATE)    // 本地缓存失效策略
    .storeMode(LocalCachedMapOptions.StoreMode.LOCAL_CACHE_REDIS); // 本地+远程

RLocalCachedMap<String, Object> cache = redisson.getLocalCachedMap("hot:data", options);
策略说明
LFU最近最少使用
LRU最近最久未使用
SOFT软引用（JVM GC 时回收）
WEAK弱引用（每次 GC 时回收）
NONE不移除
StoreMode行为
LOCAL_CACHE_REDIS本地缓存 + Redis 双写
LOCAL_CACHE仅本地缓存（不同步 Redis 变更）
REDIS_ONLY仅 Redis（禁用本地缓存）

四、RTransaction（分布式事务）

Redisson 支持在 Redis 上执行分布式事务：

RTransaction transaction = redisson.createTransaction(
    TransactionOptions.defaults()
        .timeout(30, TimeUnit.SECONDS)
        .responseTimeout(30, TimeUnit.SECONDS)
);

try {
    RMap<String, String> map = transaction.getMap("myMap");
    map.put("key1", "value1");

    RAtomicLong counter = transaction.getAtomicLong("counter");
    counter.incrementAndGet();

    transaction.commit();
} catch (Exception e) {
    transaction.rollback();
}

注意：这是 Redis 层面的事务（WATCH/MULTI/EXEC 的封装），非 ACID 关系型事务。

五、RJsonStore（JSON 存储 + 查询）

Redisson 支持对 JSON 文档进行查询（需 RedisJSON 模块）：

RJsonStore store = redisson.getJsonStore("json:store");

// 存储
store.set("$.user", new User("张三", 25));

// 路径读取
String name = store.get("$.user.name");
Integer age = store.get("$.user.age");

六、RBatch（批量操作——Pipeline）

RBatch batch = redisson.createBatch(
    BatchOptions.defaults()
        .executionMode(BatchOptions.ExecutionMode.REDIS_READ_ATOMIC)
        .responseTimeout(30, TimeUnit.SECONDS)
        .retryInterval(1000)
);

// 添加批量命令
RFuture<String> getFuture = batch.getBucket("key1").getAsync();
RFuture<Boolean> setFuture = batch.getBucket("key2").setAsync("value2");

// 一次性执行（Pipeline）
BatchResult result = batch.execute();

// 获取结果
String val1 = getFuture.get();

七、Redisson 与 Spring Cache 集成

@Configuration
@EnableCaching
public class CacheConfig {

    @Bean
    public CacheManager cacheManager(RedissonClient redisson) {
        Map<String, CacheConfig> config = new HashMap<>();

        // 用户缓存：TTL 10 分钟
        config.put("users", new CacheConfig(10 * 60 * 1000, 0));

        // 商品缓存：本地缓存 + Redis
        config.put("products", new CacheConfig(60 * 60 * 1000, 0)
            .setStoreMode(CacheConfig.StoreMode.LOCAL_CACHE_REDIS));

        // 本地缓存配置：最大 1000 个，LFU
        LocalCachedMapOptions options = LocalCachedMapOptions.defaults()
            .evictionPolicy(LocalCachedMapOptions.EvictionPolicy.LFU)
            .cacheSize(1000);
        config.put("products", new LocalCachedCacheConfig(60 * 60 * 1000, 0, options));

        return new RedissonSpringCacheManager(redisson, config);
    }
}
@Service
public class UserService {

    @Cacheable(value = "users", key = "#id")
    public User getUser(Long id) {
        // 从数据库查询
        return userRepository.findById(id);
    }

    @CachePut(value = "users", key = "#user.id")
    public User updateUser(User user) {
        return userRepository.save(user);
    }

    @CacheEvict(value = "users", key = "#id")
    public void deleteUser(Long id) {
        userRepository.deleteById(id);
    }
}

八、Redisson 与 Spring 事务集成

@Configuration
public class TransactionConfig {

    @Bean
    public TransactionManager transactionManager(RedissonClient redisson) {
        return new RedissonTransactionManager(redisson);
    }
}

@Service
public class OrderService {

    @Transactional(transactionManager = "transactionManager")
    public void createOrder(Order order) {
        // Redisson 分布式事务
        // 配合 @Transactional，保证 Redis 操作的事务性
    }
}

九、生产环境配置建议

Config config = new Config();

// 线程池配置
config.setNettyThreads(32);
config.setThreads(16);

// 序列化
config.setCodec(new Kryo5Codec());

// 连接池
config.useSingleServer()
    .setAddress("redis://localhost:6379")
    .setConnectionPoolSize(32)          // 连接池大小
    .setConnectionMinimumIdleSize(8)    // 最小空闲连接
    .setIdleConnectionTimeout(10000)    // 空闲连接超时
    .setConnectTimeout(5000)            // 连接超时
    .setTimeout(3000)                   // 命令执行超时
    .setRetryAttempts(3)                // 重试次数
    .setRetryInterval(1500)             // 重试间隔

    // 可靠监听器
    .setPingConnectionInterval(30000)   // 心跳间隔
    .setKeepPubSubOrder(true);          // 保证 Pub/Sub 顺序

// 故障转移
config.setLockWatchdogTimeout(30000);   // 锁看门狗超时

十、故障排查

问题可能原因解决方案
获取锁超时Redis 连接池满了增大连接池/检查慢查询
锁不能释放业务异常未 unlockfinally 释放/设置过期时间
数据不一致序列化导致改用 Kryo/JSON
连接断开网络问题增加重试次数 + 心跳
内存暴涨未设过期时间设置 TTL / 用 RMapCache

推荐文章

1
系统配置
Arch Linux 配置 Swap 分区并启用休眠
2
redisson
分布式锁
3
redisson
Redisson 限流与高级特性

相关文章

1
redisson
分布式锁
2
系统配置
Arch Linux 配置 Swap 分区并启用休眠