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("
");

            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;

    }

}