Redis消息监听器实现高频消息处理和防止消息丢失

本文介绍了使用Redis消息监听器实现高频消息处理，并提供了解决方案以防止消息丢失。

问题

在实际应用中，当消息频率非常高时，如果使用简单的计数器来限制处理速度，那么超过限制的消息将被直接丢弃。例如，以下代码片段展示了一个使用计数器来限制消息处理速度的Redis消息监听器：

@Slf4j
@Component
public class RedisMessageListenerListener implements MessageListener {
    private AtomicInteger counter = new AtomicInteger(0);
    private Timer timer = new Timer();
    @Resource
    private RedisTemplate<String, Object> redisTemplate;
    @Autowired
    private AsyncInstDataCapture asyncInstDataCapture;

    /**
     * 消息处理
     *
     * @param message
     * @param pattern
     */
    @Override
    public void onMessage(Message message, byte[] pattern) {
        // 增加计数器
        counter.incrementAndGet();

        // 判断计数器是否超过限制
        if (counter.get() <= 10) {
            // 处理消息
            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);

            Boolean result = redisTemplate.opsForValue().setIfAbsent(md5DigestAsHex, '1', 20, TimeUnit.SECONDS);
            if (Boolean.TRUE.equals(result)) {
                // redis消息进行处理
                log.info('接收的结果：' + deserialize.toString());
                if (StringUtil.isNotEmpty(deserialize.toString())) {
                    HashMap hashMap = JSON.parseObject(deserialize.toString(), HashMap.class);
                    String instid = hashMap.get('instid').toString();
                    String bar = hashMap.get('bar').toString();
                    switch (bar) {
                        case '1m':
                            asyncInstDataCapture.dataCapture1m(instid);
                            break;
                        case '5m':
                            asyncInstDataCapture.dataCapture5m(instid);
                            break;
                        case '15m':
                            asyncInstDataCapture.dataCapture15m(instid);
                            break;
                        case '30m':
                            asyncInstDataCapture.dataCapture30m(instid);
                            break;
                        case '1h':
                            asyncInstDataCapture.dataCapture1h(instid);
                            break;
                        case '4h':
                            asyncInstDataCapture.dataCapture4h(instid);
                            break;
                        default:
                    }
                }
            } else {
                log.info('其他服务处理中');
            }
        }

        // 定时器每秒重置计数器
        timer.schedule(new TimerTask() {
            @Override
            public void run() {
                counter.set(0);
            }
        }, 1200);
    }
}

这段代码的问题在于，如果计数器超过限制后，没有对消息进行保存或处理，而是直接丢弃了。

解决方案

为了解决这个问题，可以将超过限制的消息存储到一个队列中，然后异步处理这个队列中的消息。

1. 引入Redis队列

首先，需要引入一个队列，可以使用Redis的List数据结构来实现。在RedisMessageListenerListener类中添加以下代码：

@Autowired
private RedisTemplate<String, Object> redisTemplate;
private static final String QUEUE_KEY = 'message_queue';

2. 将消息存储到队列

然后，在计数器超过限制的判断处，将消息存储到队列中：

if (counter.get() <= 10) {
    // 处理消息
    // ...
} else {
    // 将消息存储到队列中
    redisTemplate.opsForList().rightPush(QUEUE_KEY, message);
}

3. 创建异步任务处理队列消息

接下来，可以创建一个异步任务来处理队列中的消息。在RedisMessageListenerListener类中添加以下代码：

@Async
public void processMessageQueue() {
    while (true) {
        // 从队列中取出一条消息
        Message message = (Message) redisTemplate.opsForList().leftPop(QUEUE_KEY);
        if (message == null) {
            break;
        }

        // 处理消息
        // ...
    }
}

4. 启动异步任务

最后，在启动类或配置类中，调用该异步任务：

@Autowired
private RedisMessageListenerListener redisMessageListenerListener;

@PostConstruct
public void init() {
    redisMessageListenerListener.processMessageQueue();
}

总结

通过引入Redis队列和异步处理机制，可以有效解决高频消息处理时可能出现的消息丢失问题，确保消息的完整处理。

注意：

该方案需要根据实际情况调整队列的大小和处理速度，以确保队列不会出现堆积或处理速度跟不上消息生成速度。
异步任务需要进行合理的错误处理，避免程序崩溃或出现异常。
为了提高效率，可以考虑使用多线程来处理队列中的消息。
也可以使用其他消息队列，例如Kafka或RabbitMQ，来替代Redis队列。