数据中台预警监控功能简要设计

标签：COMMENT 简要 预警 counter DEFAULT 监控 import NULL id

需求：

　　针对产品主要流程中一些异常情况， 数据中台接收各个业务系统，其他中台系统发送的异常事件。 满足触发阀值之后进行短信告警功能。

 

设计：

　　1、配置监控器：

　　　　　　监控事件、监控范围、触发阀值、对应负责人（配置负责人）

　　　　　　监控器监控的范围可以是全项目单个事件，也可以是指定项目的单个事件

　　2、某个事件连续触发阀值次数的时候，生成告警记录，发送短信。已告警的所属事件继续出发时候累计触发次数

　　3、告警记录点击处理，清除掉此记录上面累计的所有堆积事件，监控器重置，重新开始接收

 

　　大体表结构：

-- scp_alert_monitoring_configuration: 预警监控配置表
-- scp_alert_contact: 告警联系人表
-- 异常记录表 scp_business_exception_record
-- 告警计数器表 scp_alarm_counter
-- 异常记录告警计数器关联表 scp_exception_record_alarm_counter_ref
-- 预警记录表 scp_alarm_record
-- 预警短信信息表 scp_alarm_sms_record

 

设计图：

 

 

 

 

 

 

 

 

针对各个事件的数据清洗，采用策略模式分发处理

监控器的创建与维护，每一次接收新事件记录时候校验维护一次，因为配置信息不在中台数据库

数据库表sql：

CREATE TABLE `supply_chain_dsc`.`scp_alert_monitoring_configuration` (
`id` int(11) NOT NULL AUTO_INCREMENT,
`alert_event` varchar(3) DEFAULT NULL COMMENT '预警事件(01-授信决策失败、02-授信决策超时、03-额度激活失败、04-钱包开通失败、05-用信决策失败、06-用信决策超时、07-支付失败、08-客户自主还款失败)',
`monitoring_scope` varchar(2) DEFAULT NULL COMMENT '监控范围(01-单项目、02-全平台)',
`cumulative_times` int(11) DEFAULT NULL COMMENT '累计次数',
`alarm_level` varchar(2) DEFAULT NULL COMMENT '告警级别:01-A、02-B',
`is_deleted` tinyint(1) NOT NULL DEFAULT '0' COMMENT '逻辑删除(1:已删除，0:未删除)',
`create_time` datetime NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT '创建时间',
`update_time` datetime NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT '更新时间',
PRIMARY KEY (`id`)
) ENGINE=InnoDB AUTO_INCREMENT=17 DEFAULT CHARSET=utf8 ROW_FORMAT=COMPACT COMMENT='预警监控配置表';


CREATE TABLE `supply_chain_dsc`.`scp_alert_contact` (
`id` int(11) NOT NULL AUTO_INCREMENT,
`personnel_type` varchar(2) DEFAULT NULL COMMENT '人员类型(01-客户经理、02-运营经理、03-产品经理、04-风险经理)',
`name` varchar(20) DEFAULT NULL COMMENT '姓名',
`phone_no` varchar(15) NOT NULL COMMENT '手机号',
`is_deleted` tinyint(1) NOT NULL DEFAULT '0' COMMENT '逻辑删除(1:已删除，0:未删除)',
`create_time` datetime NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT '创建时间',
`update_time` datetime NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT '更新时间',
PRIMARY KEY (`id`)
) ENGINE=InnoDB AUTO_INCREMENT=2 DEFAULT CHARSET=utf8 ROW_FORMAT=COMPACT COMMENT='告警联系人表';

CREATE TABLE `supply_chain_dsc`.`scp_dsc_alarm_counter` (
`id` int(11) NOT NULL AUTO_INCREMENT,
`project_id` varchar(64) DEFAULT NULL COMMENT '项目id',
`monitoring_scope` char(2) NOT NULL COMMENT '阀值范围：01-单项目、02-全平台',
`event_type` char(2) NOT NULL COMMENT '异常类型：01-授信决策失败，02-授信决策超时，03-额度激活失败，04-钱包开通失败，05-用信决策失败，06-用信决策超时，07-支付失败，08-客户自主还款失败',
`alarm_trigger_base_times` int(11) NOT NULL COMMENT '告警触发基础次数',
`alarm_level` char(2) NOT NULL COMMENT '告警级别:01-A、02-B',
`trigger_times` int(11) NOT NULL DEFAULT '0' COMMENT '触发次数',
`alarm_status` char(1) NOT NULL DEFAULT '2' COMMENT '告警状态：1-已告警，2-未告警',
`is_deleted` tinyint(1) NOT NULL DEFAULT '0' COMMENT '''逻辑删除(1:已删除，0:未删除)''',
`reset_time` datetime NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT '重置时间',
`create_time` datetime NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT '创建时间',
`update_time` datetime NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT '更新时间',
PRIMARY KEY (`id`)
) ENGINE=InnoDB AUTO_INCREMENT=15 DEFAULT CHARSET=utf8 COMMENT='告警计数器表';
CREATE TABLE `supply_chain_dsc`.`scp_dsc_alarm_record` (
`id` int(11) NOT NULL AUTO_INCREMENT,
`project_id` varchar(64) DEFAULT NULL COMMENT '项目id',
`alarm_counter_id` int(11) NOT NULL COMMENT '计数器id',
`trigger_times` int(11) NOT NULL COMMENT '触发次数',
`monitoring_scope` char(2) NOT NULL COMMENT '监控范围(01-单项目、02-全平台)',
`event_type` char(2) NOT NULL COMMENT '预警事件(01-授信决策失败、02-授信决策超时、03-额度激活失败、04-钱包开通失败、05-用信决策失败、06-用信决策超时、07-支付失败、08-客户自主还款失败)',
`alarm_level` char(2) NOT NULL COMMENT '告警级别:01-A、02-B',
`status` char(1) NOT NULL DEFAULT '1' COMMENT '状态：1-未处理、2-已处理',
`deal_time` datetime DEFAULT NULL COMMENT '处理时间',
`deal_user_phone` varchar(255) DEFAULT NULL COMMENT '处理人手机号',
`deal_user_name` varchar(255) DEFAULT NULL COMMENT '处理人',
`create_time` datetime DEFAULT CURRENT_TIMESTAMP COMMENT '创建时间',
`update_time` datetime DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT '更新时间',
PRIMARY KEY (`id`),
KEY `index_alarm_counter_id` (`alarm_counter_id`) COMMENT '告警计数器id索引'
) ENGINE=InnoDB AUTO_INCREMENT=11 DEFAULT CHARSET=utf8 COMMENT='预警记录表';
CREATE TABLE `supply_chain_dsc`.`supply_chain_dsc`.`scp_dsc_alarm_sms_record` (
`id` int(11) NOT NULL AUTO_INCREMENT,
`alarm_record_id` int(11) NOT NULL COMMENT '告警记录id',
`name` varchar(20) DEFAULT NULL COMMENT '姓名',
`phone_no` varchar(15) DEFAULT NULL COMMENT '手机号',
`content` varchar(500) NOT NULL COMMENT '短信内容',
`create_time` datetime NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT '创建时间',
`update_time` datetime NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT '更新时间',
PRIMARY KEY (`id`),
KEY `index_alarm_record_id` (`alarm_record_id`) COMMENT '告警记录id索引'
) ENGINE=InnoDB AUTO_INCREMENT=11 DEFAULT CHARSET=utf8 COMMENT='告警短信记录表';
CREATE TABLE `supply_chain_dsc`.`scp_dsc_business_exception_record` (
`id` int(11) NOT NULL AUTO_INCREMENT,
`exception_type` char(2) NOT NULL COMMENT '异常类型：01-授信决策失败，02-授信决策超时，03-额度激活失败，04-钱包开通失败，05-用信决策失败，06-用信决策超时，07-支付失败，08-客户自主还款失败',
`business_no` varchar(100) NOT NULL COMMENT '业务单号',
`project_id` varchar(64) NOT NULL COMMENT '项目id',
`project_name` varchar(255) DEFAULT NULL COMMENT '项目名称',
`user_id` varchar(50) DEFAULT NULL COMMENT '用户id',
`user_name` varchar(100) DEFAULT NULL COMMENT '用户名称',
`fail_reason` varchar(1024) NOT NULL COMMENT '错误原因',
`fail_reason_detail` varchar(500) DEFAULT NULL COMMENT '错误详细原因',
`remark` varchar(200) DEFAULT NULL COMMENT '备注',
`trigger_time` datetime NOT NULL COMMENT '触发时间',
`create_time` datetime NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT '创建日期',
`update_time` datetime NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT '更新时间',
PRIMARY KEY (`id`)
) ENGINE=InnoDB AUTO_INCREMENT=214 DEFAULT CHARSET=utf8 COMMENT='业务异常记录表';
CREATE TABLE `supply_chain_dsc`.`scp_dsc_exception_record_alarm_counter_ref` (
`id` int(11) NOT NULL AUTO_INCREMENT,
`alarm_counter_id` int(11) NOT NULL COMMENT '告警计数器id',
`exception_record_id` int(11) NOT NULL COMMENT '异常记录id',
`alarm_record_id` int(11) DEFAULT NULL COMMENT '告警记录id',
`create_time` datetime NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT '创建时间',
`update_time` datetime NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT '更新时间',
PRIMARY KEY (`id`)
) ENGINE=InnoDB AUTO_INCREMENT=198 DEFAULT CHARSET=utf8 COMMENT='异常记录告警计数器关联表';

View Code

 

为了方便下次理解，贴主要代码：

package com.wxsbank.supplychain.dsc.business.impl;

import cn.hutool.core.collection.CollUtil;
import com.alibaba.fastjson.JSONObject;
import com.baomidou.mybatisplus.core.metadata.IPage;
import com.wxsbank.common.exception.BusinessMessage;
import com.wxsbank.framework.common.exception.impl.BusinessException;
import com.wxsbank.framework.spring.annotation.Business;
import com.wxsbank.supplychain.dsc.business.ExceptionEventCleanStrategy;
import com.wxsbank.supplychain.dsc.business.IExceptionAlarmBusiness;
import com.wxsbank.supplychain.dsc.business.SmsBusiness;
import com.wxsbank.supplychain.dsc.dto.input.AlarmQueryDTO;
import com.wxsbank.supplychain.dsc.dto.output.AlarmRepayInfoSimpleDTO;
import com.wxsbank.supplychain.dsc.entity.*;
import com.wxsbank.supplychain.dsc.enums.*;
import com.wxsbank.supplychain.dsc.event.AlarmPageEvent;
import com.wxsbank.supplychain.dsc.event.BatchClearAlarmEvent;
import com.wxsbank.supplychain.dsc.event.SmsEvent;
import com.wxsbank.supplychain.dsc.model.AlarmMessageModel;
import com.wxsbank.supplychain.dsc.model.AlarmMessagePageModel;
import com.wxsbank.supplychain.dsc.service.*;
import lombok.extern.slf4j.Slf4j;
import org.redisson.api.RLock;
import org.redisson.api.RedissonClient;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.transaction.support.TransactionSynchronizationAdapter;
import org.springframework.transaction.support.TransactionSynchronizationManager;

import javax.annotation.Resource;
import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.ReentrantLock;

/**
 * 异常告警业务处理类
 * @author zhen
 */
@Business
@Slf4j
public class ExceptionAlarmBusinessImpl implements IExceptionAlarmBusiness {

    @Resource
    private RedissonClient redissonClient;

    @Resource
    private IScpAlertMonitoringConfigurationService alertMonitoringConfigurationService;

    @Resource
    private IScpAlarmCounterService alarmCounterService;

    @Resource
    private IScpAlarmRecordService alarmRecordService;

    @Resource
    private IScpBusinessExceptionRecordService businessExceptionRecordService;

    @Resource
    private IScpExceptionRecordAlarmCounterRefService exceptionRecordAlarmCounterRefService;

    @Resource
    private IScpAlertContactService alertContactService;

    @Resource
    private IScpAlarmSmsRecordService smsRecordService;

    @Resource
    private ICreditApplyInfoService creditApplyInfoService;

    @Resource
    private IUseCreditApplyInfoService useCreditApplyInfoService;

    @Resource
    private ICreditApplyStatusRecordService creditApplyStatusRecordService;

    @Resource
    private IUseCreditApplyStatusRecordService useCreditApplyStatusRecordService;

    @Resource
    private IRepayInfoService repayInfoService;

    @Resource
    private IRmScpPclGrantCreditApplyService pclGrantCreditApplyService;

    @Resource
    private IPclUseCreditApplyInfoService pclUseCreditApplyInfoService;

    @Resource
    private SmsBusiness smsBusiness;

    @Value("${exception.monitor.alarm.sms}")
    private String monitorAlarmSmsTemplate;


    @Override
    @Transactional(rollbackFor = Exception.class)
    public void batchClearAlarm(BatchClearAlarmEvent event) {
        List<Integer> alarmRecordIdList = event.getAlarmRecordIdList();
        if (CollUtil.isEmpty(alarmRecordIdList)) {
            return;
        }
        // 判断解除人的合法性
        ScpAlertContact scpAlertContact = alertContactService.findByNameAndPhone(event.getDealUserName(), event.getDealUserPhone());
        if (scpAlertContact == null) {
            throw new BusinessException(BusinessMessage.ALERT_CONTACT_NOT_EXIST);
        }

        List<ScpAlarmRecord> scpAlarmRecordList = new ArrayList<>(alarmRecordIdList.size());
        ScpAlarmRecord scpAlarmRecord;
        ScpAlarmCounter scpAlarmCounter;
        for (Integer id : alarmRecordIdList) {
            // 告警计数器表触发次数清零
            ScpAlarmRecord alarmRecord = alarmRecordService.getById(id);
            if (alarmRecord == null || AlarmRecordStatusEnum.PROCESSED.getValue().equals(alarmRecord.getStatus())) {
                continue;
            }
            scpAlarmCounter = new ScpAlarmCounter();
            scpAlarmCounter.setId(alarmRecord.getAlarmCounterId());
            scpAlarmCounter.setTriggerTimes(0);
            scpAlarmCounter.setAlarmStatus(AlarmStatusEnum.UN_COMPLETE_ALARM.getValue());
            scpAlarmCounter.setResetTime(LocalDateTime.now());
            alarmCounterService.updateById(scpAlarmCounter);
            // 预警记录
            scpAlarmRecord = new ScpAlarmRecord();
            scpAlarmRecord.setId(id);
            // 预警记录已被处理
            scpAlarmRecord.setStatus(AlarmRecordStatusEnum.PROCESSED.getValue());
            scpAlarmRecord.setDealTime(LocalDateTime.now());
            scpAlarmRecord.setDealUserName(event.getDealUserName());
            scpAlarmRecord.setDealUserPhone(event.getDealUserPhone());
            scpAlarmRecordList.add(scpAlarmRecord);
        }
        // 预警记录表更新处理人信息
        alarmRecordService.updateBatchById(scpAlarmRecordList);
    }

    @Override
    public void clearAndSaveBusinessException(JSONObject content, ExceptionEventCleanStrategy cleanStrategy, String businessNo) {

        final String keyPrefix = "EXCEPTION_EVENT_TRIGGER_";
        String key = keyPrefix + businessNo + "_" + cleanStrategy.getExceptionEventType().getValue();
        RLock lock = redissonClient.getLock(key);
        try {
            lock.lock(2, TimeUnit.MINUTES);
            //清洗数据
            Integer exceptionRecordId = cleanStrategy.cleanAndSave(content);
            if (exceptionRecordId == -1) {
                // -1 表示清洗数据出现数据异常
                return;
            }
            if (exceptionRecordId != null){

                //创建或销毁计数器
                repairCounterInfo(exceptionRecordId);

                //异步触发计数器
                Integer finalExceptionRecordId = exceptionRecordId;
                CompletableFuture.runAsync(() ->
                        triggerAlarmMonitorChain(finalExceptionRecordId)
                );

            }
        } catch (Exception e) {
            log.error("【LEVEL:LOW】【业务异常数据清洗 type:" + cleanStrategy.getExceptionEventType().getDesc() + " data:" + content.toJSONString()
                    + "】数据异常！", e);
        } finally {
            lock.unlock();
        }

    }

    /**
     * 触发报警监控链
     * @param businessExceptionRecordId
     */
    @Transactional
    @Override
    public void triggerAlarmMonitorChain(Integer businessExceptionRecordId) {
        ScpBusinessExceptionRecord exceptionRecord = businessExceptionRecordService.getById(businessExceptionRecordId);
        if (exceptionRecord == null) {
            return;
        }

        List<ScpAlarmCounter> counters = queryTriggerCounter(exceptionRecord);
        if(counters.isEmpty()){
            return;
        }

        for(ScpAlarmCounter counter : counters){
            String key = "EXCEPTION_EVENT_COUNTER_TRIGGER_" + counter.getId();
            // 增加一个同步锁，处理针对同一counter累加的并发问题
            RLock lock = redissonClient.getLock(key);
            try {
                lock.lock(2, TimeUnit.MINUTES);
                if (AlarmStatusEnum.COMPLETED_ALARM.getValue().equals(counter.getAlarmStatus())){
                    triggerInCompletedAlarmCounter(counter, businessExceptionRecordId);
                }else {
                    triggerInUnCompletedAlarmCounter(counter, businessExceptionRecordId);
                }
            }catch(Exception e){
                log.error("【LEVEL:LOW】【预警监控触发异常 counterId:" + counter.getId() + " businessExceptionRecordId:" + businessExceptionRecordId
                        + " 事件件类型：" + ExceptionEventTypeEnum.getDescByValue(counter.getEventType()) + "】数据异常！", e);
            }finally {
                if (lock.isLocked()){
                    lock.unlock();
                }
            }

        }

    }

    @Override
    public void sendAlarmSms(Integer alarmRecordId) {
        List<ScpAlarmSmsRecord> smsRecordList = smsRecordService.queryByAlarmRecordId(alarmRecordId);
        for (ScpAlarmSmsRecord alarmSmsRecord : smsRecordList){
            SmsEvent event = new SmsEvent();
            event.setContent(alarmSmsRecord.getContent());
            event.setPhone(alarmSmsRecord.getPhoneNo());
            smsBusiness.customMsgSend(event);
        }

    }

    @Override
    public List<UseCreditApplyInfo> selectDelayAndNoTriggerUseCreditInfoList(LocalDateTime startTime, LocalDateTime endTime,
                                                                             LocalDateTime maxCreateTime) {
        return useCreditApplyInfoService.selectDelayAndNoTriggerUseCreditInfoList(startTime, endTime, maxCreateTime);
    }

    @Override
    public List<PclUseCreditApplyInfo> selectDelayAndNoTriggerPersonUseCreditApplyInfoList(LocalDateTime startTime,
                                                                                           LocalDateTime endTime,
                                                                                           LocalDateTime maxCreateTime) {
        return pclUseCreditApplyInfoService.selectDelayAndNoTriggerPersonUseCreditApplyInfoList(startTime, endTime, maxCreateTime);
    }

    @Override
    public List<CreditApplyInfo> selectDelayAndNoTriggerCreditApplyInfoList(LocalDateTime startTime, LocalDateTime endTime,
                                                                            LocalDateTime maxCreateTime) {
        return creditApplyInfoService.selectDelayAndNoTriggerCreditApplyInfoList(startTime, endTime, maxCreateTime);
    }

    @Override
    public List<RmScpPclGrantCreditApply> selectDelayAndNoTriggerPersonCreditApplyInfoList(LocalDateTime startTime,
                                                                                           LocalDateTime endTime,
                                                                                           LocalDateTime maxCreateTime) {
        return pclGrantCreditApplyService.selectDelayAndNoTriggerPersonCreditApplyInfoList(startTime, endTime, maxCreateTime);
    }

    @Override
    public List<CreditApplyStatusRecord> selectFailAndNoTriggerCreditApplyStatusList(LocalDateTime startTime, LocalDateTime endTime) {
        return creditApplyStatusRecordService.selectFailAndNoTriggerCreditApplyStatusList(startTime, endTime);
    }

    @Override
    public List<UseCreditApplyStatusRecord> selectFailAndNoTriggerUseCreditApplyStatusList(LocalDateTime startTime, LocalDateTime endTime) {
        return useCreditApplyStatusRecordService.selectFailAndNoTriggerUseCreditApplyStatusList(startTime, endTime);
    }

    @Override
    public List<AlarmRepayInfoSimpleDTO> selectFailAndNoTriggerRepayInfoList(LocalDateTime startTime, LocalDateTime endTime) {
        return repayInfoService.selectFailAndNoTriggerRepayInfoList(startTime, endTime);
    }

    @Override
    public AlarmMessagePageModel queryAlarmPage(AlarmPageEvent event) {
        AlarmQueryDTO alarmQueryDTO = new AlarmQueryDTO();
        alarmQueryDTO.setName(event.getName());
        alarmQueryDTO.setPhoneNo(event.getPhoneNo());
        alarmQueryDTO.setProjectId(event.getProjectId());
        alarmQueryDTO.setPageNum(event.getPageNum());
        alarmQueryDTO.setPageSize(event.getPageSize());
        // 查询未处理的告警记录
        alarmQueryDTO.setStatus("1");
        IPage<ScpAlarmRecord> scpAlarmRecordIPage = alarmRecordService.queryAlarmRecordByCondition(alarmQueryDTO);
        List<ScpAlarmRecord> scpAlarmRecords = scpAlarmRecordIPage.getRecords();
        if (CollUtil.isEmpty(scpAlarmRecords)) {
            return null;
        }
        AlarmMessagePageModel alarmMessagePageModel = new AlarmMessagePageModel();
        List<AlarmMessageModel> alarmMessageModelList = new ArrayList<>(scpAlarmRecords.size());
        AlarmMessageModel alarmMessageModel;
        for (ScpAlarmRecord scpAlarmRecord : scpAlarmRecords) {
            alarmMessageModel = new AlarmMessageModel();
            alarmMessageModel.setAlarmRecordId(scpAlarmRecord.getId());
            alarmMessageModel.setMonitoringScope(scpAlarmRecord.getMonitoringScope());
            alarmMessageModel.setEventType(scpAlarmRecord.getEventType());
            alarmMessageModel.setAlarmLevel(scpAlarmRecord.getAlarmLevel());
            alarmMessageModel.setTriggerTime(
                    scpAlarmRecord.getCreateTime().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss")));
            alarmMessageModel.setProjectId(scpAlarmRecord.getProjectId());
            alarmMessageModel.setProjectName(scpAlarmRecord.getProjectName());
            alarmMessageModelList.add(alarmMessageModel);
        }
        alarmMessagePageModel.setTotalNum((int) scpAlarmRecordIPage.getPages());
        alarmMessagePageModel.setTotalSize((int) scpAlarmRecordIPage.getTotal());
        alarmMessagePageModel.setAlarmMessageModelList(alarmMessageModelList);
        return alarmMessagePageModel;
    }

    /**
     * 在未完成的计数器触发
     * @param counter
     * @param businessExceptionRecordId
     */
    private void triggerInUnCompletedAlarmCounter(ScpAlarmCounter counter, Integer businessExceptionRecordId){
        int nowTriggerTimes = counter.getTriggerTimes() +1;

        //计数器触发次数+1
        ScpAlarmCounter updateCounter = new ScpAlarmCounter();
        updateCounter.setId(counter.getId());
        updateCounter.setTriggerTimes(counter.getTriggerTimes() + 1);
        alarmCounterService.updateById(updateCounter);

        //异常数据-触发器关联记录落库
        ScpExceptionRecordAlarmCounterRef ref = new ScpExceptionRecordAlarmCounterRef();
        ref.setAlarmCounterId(counter.getId());
        ref.setExceptionRecordId(businessExceptionRecordId);
        exceptionRecordAlarmCounterRefService.save(ref);

        //触发告警
        if(nowTriggerTimes == counter.getAlarmTriggerBaseTimes()){
            //触发器状态变更已触发
            updateCounter = new ScpAlarmCounter();
            updateCounter.setId(counter.getId());
            updateCounter.setAlarmStatus(AlarmStatusEnum.COMPLETED_ALARM.getValue());
            alarmCounterService.updateById(updateCounter);

            ScpBusinessExceptionRecord businessExceptionRecord = businessExceptionRecordService.getById(businessExceptionRecordId);
            //创建告警记录
            ScpAlarmRecord alarmRecord = new ScpAlarmRecord();
            alarmRecord.setTriggerTimes(nowTriggerTimes);
            alarmRecord.setAlarmCounterId(counter.getId());
            alarmRecord.setProjectId(counter.getProjectId());
            alarmRecord.setEventType(counter.getEventType());
            alarmRecord.setAlarmLevel(counter.getAlarmLevel());
            alarmRecord.setMonitoringScope(counter.getMonitoringScope());
            alarmRecord.setStatus(AlarmRecordStatusEnum.UNTREATED.getValue());
            alarmRecord.setProjectName(businessExceptionRecord.getProjectName());
            alarmRecordService.eachIdSave(alarmRecord);

            //为计数器内包含的时间关联与告警记录创建关联
            List<ScpExceptionRecordAlarmCounterRef> refList
                    = exceptionRecordAlarmCounterRefService.selectListByCurrentCounter(counter);
            refList.forEach(item->
                    item.setAlarmRecordId(alarmRecord.getId())
            );
            exceptionRecordAlarmCounterRefService.updateBatchById(refList);

            //目前是发送给所有的联系人
            List<ScpAlertContact> contactList = alertContactService.queryValidContactList();
            List<ScpAlarmSmsRecord> smsSendList = new ArrayList<>();
            if(!contactList.isEmpty()){
                for(ScpAlertContact  contact : contactList){
                    //创建smsRecord记录
                    ScpAlarmSmsRecord scpAlarmSmsRecord = new ScpAlarmSmsRecord();
                    scpAlarmSmsRecord.setName(contact.getName());
                    scpAlarmSmsRecord.setPhoneNo(contact.getPhoneNo());
                    scpAlarmSmsRecord.setAlarmRecordId(alarmRecord.getId());
                    String content = String.format(monitorAlarmSmsTemplate, AlarmLevelEnum.getDescByValue(alarmRecord.getAlarmLevel()),
                            MonitorScopeEnum.getDescByValue(alarmRecord.getMonitoringScope())
                                    + (MonitorScopeEnum.SINGLE_PROJECT.getValue().equals(alarmRecord.getMonitoringScope()) ?
                                    "(" + alarmRecord.getProjectName() + ")" : "")
                                    + " " + ExceptionEventTypeEnum.getDescByValue(alarmRecord.getEventType()));
                    scpAlarmSmsRecord.setContent(content);
                    smsSendList.add(scpAlarmSmsRecord);
                }
                //告警记录绑定发送短信信息落库
                smsRecordService.saveBatch(smsSendList);

                //异步触发发告警短信
                if(TransactionSynchronizationManager.isActualTransactionActive()){
                    //如果事务未提交，则提交之后调用异步方法
                    TransactionSynchronizationManager.registerSynchronization(new TransactionSynchronizationAdapter() {
                        @Override
                        public void afterCommit() {
                            CompletableFuture.runAsync(() ->
                                    sendAlarmSms(alarmRecord.getId())
                            );
                        }
                    });
                }else {
                    CompletableFuture.runAsync(() ->
                            sendAlarmSms(alarmRecord.getId())
                    );

                }

            }


        }


    }

    /**
     * 在已完成的计数器触发
     * @param counter
     * @param businessExceptionRecordId
     */
    private void triggerInCompletedAlarmCounter(ScpAlarmCounter counter, Integer businessExceptionRecordId){
        ScpAlarmCounter updateCounter = new ScpAlarmCounter();
        updateCounter.setId(counter.getId());
        updateCounter.setTriggerTimes(counter.getTriggerTimes() + 1);
        alarmCounterService.updateById(updateCounter);
        ScpAlarmRecord alarmRecord = alarmRecordService.findUntreatedRecordByCounter(counter.getId());

        ScpAlarmRecord updateAlarmRecord = new ScpAlarmRecord();
        updateAlarmRecord.setId(alarmRecord.getId());
        updateAlarmRecord.setTriggerTimes(alarmRecord.getTriggerTimes() + 1);
        alarmRecordService.updateById(updateAlarmRecord);

        ScpExceptionRecordAlarmCounterRef ref = new ScpExceptionRecordAlarmCounterRef();
        ref.setAlarmCounterId(counter.getId());
        ref.setAlarmRecordId(alarmRecord.getId());
        ref.setExceptionRecordId(businessExceptionRecordId);
        exceptionRecordAlarmCounterRefService.save(ref);
    }

    /**
     * 查询将要触发的counter
     * @param exceptionRecord
     * @return
     */
    private List<ScpAlarmCounter> queryTriggerCounter(ScpBusinessExceptionRecord exceptionRecord) {

        List<ScpAlarmCounter> counterList = alarmCounterService.queryWaitTriggerCounterList(exceptionRecord.getExceptionType(),
                MonitorScopeEnum.ALL_PROJECT.getValue(), null);

        List<ScpAlarmCounter> singleProjectCounterList
                = alarmCounterService.queryWaitTriggerCounterList(exceptionRecord.getExceptionType(),
                MonitorScopeEnum.SINGLE_PROJECT.getValue(), exceptionRecord.getProjectId());

        counterList.addAll(singleProjectCounterList);
        return counterList;
    }

    /**
     * 维护计数器
     * @param exceptionRecordId  异常事件id
     */
    private void repairCounterInfo(Integer exceptionRecordId){

        // 1、找到应该删除的counter，完成删除
        List<Integer> counterIds = alarmCounterService.queryShouldInvalidCounterIds();
        if (!counterIds.isEmpty()) {
            alarmCounterService.invalidCounter(counterIds);
        }

        String key = "EXCEPTION_EVENT_TRIGGER_ADD_COUNTER";
        RLock lock = redissonClient.getLock(key);
        try {
            lock.lock(2, TimeUnit.MINUTES);

            // 2、找到应该新建的counter，完成新建
            ScpBusinessExceptionRecord exceptionRecord = businessExceptionRecordService.getById(exceptionRecordId);
            String eventType = exceptionRecord.getExceptionType();
            String projectId = exceptionRecord.getProjectId();
            List<ScpAlertMonitoringConfiguration> allScopeConfigurations
                    = alertMonitoringConfigurationService.queryByEventAndScope(eventType, MonitorScopeEnum.ALL_PROJECT.getValue());
            List<ScpAlertMonitoringConfiguration> singleProjectConfigurations
                    = alertMonitoringConfigurationService.queryByEventAndScope(eventType, MonitorScopeEnum.SINGLE_PROJECT.getValue());

            for (ScpAlertMonitoringConfiguration configuration : allScopeConfigurations) {
                ScpAlarmCounter counter = alarmCounterService.findAllProjectMatchValidCounter(configuration);
                if (counter == null) {
                    createNewCounter(configuration, null);
                }
            }

            for (ScpAlertMonitoringConfiguration configuration : singleProjectConfigurations) {
                ScpAlarmCounter counter = alarmCounterService.findSingleProjectMatchValidCounter(configuration, projectId);
                if (counter == null) {
                    createNewCounter(configuration, projectId);
                }
            }
        }catch(Exception e) {
            log.error("【LEVEL:LOW】【维护计数器 新增计数器异常】", e);
            throw e;
        }finally {
            if(lock.isLocked()){
                lock.unlock();
            }
        }

    }


    /**
     * 创建新的计数器
     * @param configuration 配置器
     * @param projectId  项目id
     */
    private void createNewCounter(ScpAlertMonitoringConfiguration configuration, String projectId){
        ScpAlarmCounter counter = new ScpAlarmCounter();
        counter.setAlarmLevel(configuration.getAlarmLevel());
        counter.setMonitoringScope(configuration.getMonitoringScope());
        counter.setEventType(configuration.getAlertEvent());
        counter.setAlarmTriggerBaseTimes(configuration.getCumulativeTimes());
        counter.setAlarmStatus(AlarmStatusEnum.UN_COMPLETE_ALARM.getValue());
        counter.setProjectId(projectId);
        alarmCounterService.save(counter);
    }


}

View Code

 

标签：COMMENT,简要,预警,counter,DEFAULT,监控,import,NULL,id
From： https://www.cnblogs.com/aigeileshei/p/16955633.html