Redis 消息监听器: 使用 @Slf4j 和 @Component 注解实现消息处理

@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);
}

}