Kafka+redis分布式锁结合使用心得总结

#kafka部分

@KafkaListener(topics = "#{'${vsmart_alert_detection_tms_send_message_topic}'.split(',')}", groupId = "${vsmart.alert.detection.consumer.group}")
public void vsmartAlertDetectionTmsSendMessage(ConsumerRecord<?, ?> record, Acknowledgment ack, @Header(KafkaHeaders.RECEIVED_TOPIC) String topic) {
    doSendMessage(record,ack);
}

private void doSendMessage(ConsumerRecord<?, ?> record, Acknowledgment ack) {
    Optional message = Optional.ofNullable(record.value());
    String key = record.topic() + "-" + record.partition() + "-offset:" + record.offset();
    if (RedisUtils.isExistsKey(key)) {
        ack.acknowledge();
        return;
    }

    try {
        if (message.isPresent() && (record.timestamp() > (System.currentTimeMillis() - kafkaConsumerDelayTime))) {
            JSONObject msg = JSONObject.parseObject(record.value().toString());
            msg.put(VSMART_KAFKA_MSG_POSITION_INFO, key);

            //具体操作
        }
    }catch (Exception e){
        
    }finally {
        ack.acknowledge();
    }
}    

#redis部分 

public Boolean handler(JSONObject msg) {
    //解析
    Boolean isOk = jsonToDetectionInfos(msg);

    if (!isOk) {
        return false;
    }

    //加锁 associatedKey()
    
    String lockKey = associatedKey();
    if (StrUtil.isEmpty(lockKey)) {
        return false;
    }
    RLock lock = SpringUtils.getBean(RedissonClient.class).getLock(lockKey);
    //锁的时间 根据业务需要进行调整
    try {
        boolean flag_2 = lock.tryLock(10, 300, TimeUnit.SECONDS);

        if (flag_2) {        
            //加锁后执行前判断是否已经处理过kafka中相同位置的信息了
            if (ObjectUtil.isNotNull(msg) &&
                    ObjectUtil.isNotNull(msg.get(VSMART_KAFKA_MSG_POSITION_INFO)) &&
                    RedisUtils.isExistsKey(msg.getString(VSMART_KAFKA_MSG_POSITION_INFO))) {            
                return false;
            }

            //具体业务操作
            //...
                
            return true;
        } else {
            detectionRuleBo.getLogText().append(StrUtil.format("{}-获取锁失败;", detectionRuleBo.getName())).append("<br>");
            return false;
        }

    } catch (Exception e) {
        
    } finally {
        ///释放锁
        if (null != lock && lock.isHeldByCurrentThread()) {        
            if (ObjectUtil.isNotNull(msg) &&
                    ObjectUtil.isNotNull(msg.get(VSMART_KAFKA_MSG_POSITION_INFO))) {
                RedisUtils.setCacheStrExpire(msg.getString(VSMART_KAFKA_MSG_POSITION_INFO), msg.getString(VSMART_KAFKA_MSG_POSITION_INFO), 60 * 60);
            }
            //解锁
            lock.unlock();
        }
        return true;
    }
}