pinpoint web报警机制源码解析

<!--定义了一个alramJob的批处理任务-->
　　<batch:job id="alarmJob">    
　　　　<batch:step id="alarmPartitionStep">        
　　　　　　<!--此alarmJob只有一个Step-->
        <batch:partition step="alarmStep" partitioner="alarmPartitioner">
            <!--设置执行的线程池-->
            <batch:handler task-executor="alarmPoolTaskExecutorForPartition" />
        </batch:partition>
    </batch:step>
    <batch:listeners>
       <batch:listener ref="jobFailListener"/>
    </batch:listeners>
</batch:job>

<batch:step id="alarmStep">
    <!--代表step的一种处理策略-->
    <batch:tasklet>
        <!--批处理流程-->
        <!-- 顺序执行
        reader:读取数据 => 提供Checker，即异常校验器，见下面的bean
        processor:处理数据 => 用Checker进行校验，见下面的bean
        writer:回写数据 => 判断Checker是否有异常情况，有则报警，见下面的bean
        -->
        <batch:chunk reader="reader" processor="processor" writer="writer" commit-interval="1"/>
    </batch:tasklet>
</batch:step>

<bean id="alarmPartitioner" class="com.navercorp.pinpoint.web.alarm.AlarmPartitioner"/>

<bean id="reader" class="com.navercorp.pinpoint.web.alarm.AlarmReader" scope="step"/>
<bean id="processor" class="com.navercorp.pinpoint.web.alarm.AlarmProcessor" scope="step"/>
<bean id="writer" class="com.navercorp.pinpoint.web.alarm.AlarmWriter" scope="step"/>

// StepExecutionListener监听器，可以定义Step开始前后的操作
public class AlarmReader implements ItemReader<AlarmChecker>, StepExecutionListener {
    ...
　　 // 报警所用的Checker在内存里
    private final Queue<AlarmChecker> checkers = new ConcurrentLinkedDeque<>();
    ...

    // Checker出队，供processor使用，
　　 @Override
    public AlarmChecker read() {
        return checkers.poll();
    }

    // 批处理之前，将应用的报警规则加入Checker之中
    @Override
    public void beforeStep(StepExecution stepExecution) {
　　　　　// 查询所有的应用
        List<Application> applicationList = applicationIndexDao.selectAllApplicationNames();
　　　　 // 根据应用用户配置的规则，添加Checker到队列当中
        for (Application application : applicationList) {
            addChecker(application);
        }
    }

    private void addChecker(Application application) {
　　　　 // 根据应用名称获取所有的规则，应用名称就是配置agent的时候，指定的applicationName
        List<Rule> rules = alarmService.selectRuleByApplicationId(application.getName());
        long timeSlotEndTime = System.currentTimeMillis();
        Map<DataCollectorCategory, DataCollector> collectorMap = new HashMap<>();
        // 遍历规则
        for (Rule rule : rules) {
　　　　　　  // CheckerCategory是一个枚举类，预置了所有的报警规则模版，比如失败请求次数、慢请求次数等
            CheckerCategory checkerCategory = CheckerCategory.getValue(rule.getCheckerName());
　　　　　　  // 数据收集器是为检验规则准备的，例如Rule是失败请求次数，但是次数从哪里来，就是从这个收集器来的

            DataCollector collector = collectorMap.get(checkerCategory.getDataCollectorCategory());
　　　　　　　// 这里是一个基于Map的缓存
            if (collector == null) {
                collector = dataCollectorFactory.createDataCollector(checkerCategory, application, timeSlotEndTime);
                collectorMap.put(collector.getDataCollectorCategory(), collector);
            }
            // 创建Checker，有兴趣的读者可以看看CheckerCategroy的源码，设计的还是很不错的。
　　　　　　  // AlaramChecker是一个抽象方法，具体的功能由子类实现
            AlarmChecker checker = checkerCategory.createChecker(collector, rule);
　　　　　　　// 加入队列
            checkers.add(checker);
        }
        
    }
    ...
}

public class AlarmProcessor implements ItemProcessor<AlarmChecker, AlarmChecker> {
　　 // 此处的AlarmChecker是上面的read()方法传递过来的
    @Override
    public AlarmChecker process(AlarmChecker checker) {
        // check，顾名思义，检验，标记是否有异常情况,check()方法见下
        checker.check();
        return checker;
    }

}

public class AlarmWriter implements ItemWriter<AlarmChecker> {

    // 需要用户自定义配置在Spring中的AlarmMessageSender，如果不配置，则是一个空实现
    @Autowired(required = false)
    private AlarmMessageSender alarmMessageSender = new EmptyMessageSender();

    @Autowired
    private AlarmService alarmService;

　　 // 实现的接口的方法，主要内容
    @Override
    public void write(List<? extends AlarmChecker> checkers) throws Exception {
        Map<String, CheckerResult> beforeCheckerResults = alarmService.selectBeforeCheckerResults(checkers.get(0).getRule().getApplicationId());
        // 遍历上面传递的Checker
        for (AlarmChecker checker : checkers) {
            CheckerResult beforeCheckerResult = beforeCheckerResults.get(checker.getRule().getCheckerName());

            if (beforeCheckerResult == null) {
                beforeCheckerResult = new CheckerResult(checker.getRule().getApplicationId(), checker.getRule().getCheckerName(), false, 0, 1);
            }
            // 对上面的Processor标记的detected值进行检查
            if (checker.isDetected()) {
                sendAlarmMessage(beforeCheckerResult, checker);
            }
            // 记录报警历史
            alarmService.updateBeforeCheckerResult(beforeCheckerResult, checker);
        }
    }

    private void sendAlarmMessage(CheckerResult beforeCheckerResult, AlarmChecker checker) {
        if (isTurnToSendAlarm(beforeCheckerResult)) {
            // 是否配置了发送报警短信
            if (checker.isSMSSend()) {
                alarmMessageSender.sendSms(checker, beforeCheckerResult.getSequenceCount() + 1);
            }
            // 是否配置了发送报警邮件
            if (checker.isEmailSend()) {
                alarmMessageSender.sendEmail(checker, beforeCheckerResult.getSequenceCount() + 1);
            }
        }

    }
　　// ...
}