Redis消息监听器实现 - 使用@Slf4j注解和异步数据捕获

@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中正确配置了消息队列，并且消息发送方和接收方都连接到了同一个Redis实例。

2. 消息处理过程中发生了异常。在处理消息的过程中，如果发生了异常，可能导致消息丢失。需要检查消息处理的逻辑，并确保没有发生异常。

3. 消息处理时间过长。如果消息处理时间过长，可能会导致其他消息被覆盖或丢失。可以考虑将消息处理逻辑进行优化，以减少处理时间。

4. Redis消息队列容量不足。如果Redis消息队列容量不足，可能会导致消息被丢弃。可以检查Redis的配置，并根据需要增加队列的容量。

建议对代码进行调试和日志输出，以确定消息丢失的具体原因。可以通过打印日志来追踪消息的发送和接收过程，以及消息处理的时间和结果。同时，建议对消息队列的配置进行检查，确保配置正确并足够满足需求。