Redis 消息消费限流：每秒只消费10条数据

在使用 Redis 接收和处理消息时，如果消息量很大，可能会导致系统过载。为了避免这种情况，可以采用限流策略来限制每秒消费的消息数量。

本文将介绍如何使用 Redis 计数器和限流算法，限制每秒只消费 10 条 Redis 消息。

代码实现

首先，我们需要在 onMessage 方法中添加代码来实现限流功能。具体实现如下：

@Override
public void onMessage(Message message, byte[] pattern) {
    String channel = new String(pattern);
    log.info('onMessage --> 消息通道是：' + channel);

    RedisSerializer<?> valueSerializer = redisTemplate.getValueSerializer();
    Object deserialize = valueSerializer.deserialize(message.getBody());
    log.info('反序列化的结果:' + deserialize);
    if (deserialize == null) return;
    String md5DigestAsHex = DigestUtils.md5DigestAsHex(deserialize.toString().getBytes(StandardCharsets.UTF_8));
    log.info('计算得到的key: ' + md5DigestAsHex);

    // 获取当前秒级别的时间戳
    long currentTimestamp = System.currentTimeMillis() / 1000;

    // 自增计数器并设置过期时间为1秒
    Long count = redisTemplate.opsForValue().increment(md5DigestAsHex, 1);
    redisTemplate.expire(md5DigestAsHex, 1, TimeUnit.SECONDS);

    if (count > 10) {
        log.info('每秒只允许消费10条数据，已达到限制，拒绝处理该消息');
        return;
    }

    Boolean result = redisTemplate.opsForValue().setIfAbsent(md5DigestAsHex, '1', 20, TimeUnit.SECONDS);
    if (Boolean.TRUE.equals(result)) {
        // redis消息进行处理
        log.info('接收的结果：' + deserialize.toString());
    } else {
        log.info('其他服务处理中');
    }
}

代码解释

获取当前秒级别的时间戳： long currentTimestamp = System.currentTimeMillis() / 1000;
自增计数器并设置过期时间为1秒： Long count = redisTemplate.opsForValue().increment(md5DigestAsHex, 1); 和 redisTemplate.expire(md5DigestAsHex, 1, TimeUnit.SECONDS);。这里使用 md5DigestAsHex 作为计数器的 key，确保每个消息对应唯一的计数器。
判断计数器是否超过限制： if (count > 10)。如果超过限制，则拒绝处理该消息。
处理消息： 如果计数器没有超过限制，则可以继续处理该消息。

总结

通过使用 Redis 计数器和限流算法，我们可以有效地限制每秒消费的消息数量，防止系统被过载。同时，还可以根据实际情况调整限制阈值，以达到最佳的性能和稳定性。

希望本文能对您有所帮助！如果您有任何问题或建议，请随时留言。