Redis 消息监听器 - 基于 RedisTemplate 和 AsyncInstDataCapture 的消息处理

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

/**
 * 消息处理
 *
 * @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 {
        // 休眠1500毫秒，处理完10条数据后休眠一段时间
        Thread.sleep(1500);
    } catch (InterruptedException e) {
        log.error('线程休眠异常：' + e.getMessage());
        Thread.currentThread().interrupt();
    }
}

}