Redis消息监听器实现 - 异步数据捕获

@Slf4j @Component public class RedisMessageListenerListener implements MessageListener { @Resource private RedisTemplate<String, Object> redisTemplate; @Autowired private AsyncInstDataCapture asyncInstDataCapture;

private AtomicInteger counter = new AtomicInteger(0);

/**
 * 消息处理
 *
 * @param message
 * @param pattern
 */
@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);

    Boolean result = redisTemplate.opsForValue().setIfAbsent(md5DigestAsHex, '1', 20, TimeUnit.SECONDS);
    if (Boolean.TRUE.equals(result)) {
        // redis消息进行处理
        log.info('接收的结果：', 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('其他服务处理中');
    }

    try {
        // 等待200毫秒
        Thread.sleep(200);
    } catch (InterruptedException e) {
        // 处理异常
        // ...
        log.error('休眠时发生异常', e);
    }

}
// 针对休眠时间问题，增加了日志记录，方便排查问题
// 可以根据实际情况调整休眠时间和异常处理逻辑

}