序

本文主要研究一下PowerJob的DatabaseMonitorAspect

DatabaseMonitorAspect

tech/powerjob/server/persistence/monitor/DatabaseMonitorAspect.java

@Slf4j
@Aspect
@Component
@RequiredArgsConstructor
public class DatabaseMonitorAspect {

    private final MonitorService monitorService;

    @Around("execution(* tech.powerjob.server.persistence.remote.repository..*.*(..))")
    public Object monitorCoreDB(ProceedingJoinPoint joinPoint) throws Throwable {
        return wrapperMonitor(joinPoint, DatabaseType.CORE);
    }

    @Around("execution(* tech.powerjob.server.persistence.local..*.*(..))")
    public Object monitorLocalDB(ProceedingJoinPoint joinPoint) throws Throwable {
        return wrapperMonitor(joinPoint, DatabaseType.LOCAL);
    }

    //......
}    

DatabaseMonitorAspect拦截了remote和local的repository，然后执行wrapperMonitor

wrapperMonitor

    private Object wrapperMonitor(ProceedingJoinPoint point, DatabaseType type) throws Throwable {
        String classNameMini = AOPUtils.parseRealClassName(point);
        final String methodName = point.getSignature().getName();

        DatabaseEvent event = new DatabaseEvent().setType(type)
                .setServiceName(classNameMini)
                .setMethodName(methodName)
                .setStatus(DatabaseEvent.Status.SUCCESS);

        long startTs = System.currentTimeMillis();
        try {
            final Object ret = point.proceed();
            event.setRows(parseEffectRows(ret));
            return ret;
        } catch (Throwable t) {
            event.setErrorMsg(t.getClass().getSimpleName()).setStatus(DatabaseEvent.Status.FAILED);
            throw t;
        } finally {
            long cost = System.currentTimeMillis() - startTs;
            monitorService.monitor(event.setCost(cost));
        }
    }

wrapperMonitor主要是构建DatabaseEvent，成功时获取effectRows，异常时获取erroMsg，最后计算耗时，执行monitorService.monitor

MonitorService

tech/powerjob/server/monitor/MonitorService.java

public interface MonitorService {
    void monitor(Event event);
}

MonitorService定义了monitor方法

PowerJobMonitorService

tech/powerjob/server/monitor/PowerJobMonitorService.java

@Slf4j
@Component
public class PowerJobMonitorService implements MonitorService {

    private final List<Monitor> monitors = Lists.newLinkedList();

    public PowerJobMonitorService(List<Monitor> monitors) {
        monitors.forEach(m -> {
            log.info("[MonitorService] register monitor: {}", m.getClass().getName());
            this.monitors.add(m);
        });
    }

    @Override
    public void monitor(Event event) {
        monitors.forEach(m -> m.record(event));
    }
}

PowerJobMonitorService是实现了MonitorService接口，其monitor方法遍历monitors，挨个执行record方法

Monitor

tech/powerjob/server/monitor/Monitor.java

public interface Monitor {

    /**
     * 全局上下文绑定 & 初始化
     */
    void init();
    /**
     * 记录监控事件
     * 请注意该方法务必异步不阻塞！！！
     * @param event 事件
     */
    void record(Event event);
}

Monitor接口定义了record方法

LogMonitor

tech/powerjob/server/monitor/monitors/LogMonitor.java

@Component
public class LogMonitor implements Monitor, ServerInfoAware {

    /**
     * server 启动依赖 DB，DB会被 monitor，因此最初的几条 log serverInfo 一定为空，在此处简单防空
     */
    private ServerInfo serverInfo = new ServerInfo();

    private static final String MDC_KEY_SERVER_ID = "serverId";


    @Override
    public void init() {
    }

    @Override
    public void record(Event event) {
        MDC.put(MDC_KEY_SERVER_ID, String.valueOf(serverInfo.getId()));
        LoggerFactory.getLogger(event.type()).info(event.message());
    }

    @Override
    public void setServerInfo(ServerInfo serverInfo) {
        this.serverInfo = serverInfo;
    }
}

LogMonitor实现了Monitor, ServerInfoAware接口，其record方法通过logger打印event的message

小结

DatabaseMonitorAspect拦截了remote和local的repository，然后执行wrapperMonitor；wrapperMonitor主要是构建DatabaseEvent，成功时获取effectRows，异常时获取erroMsg，最后计算耗时，执行monitorService.monitor；LogMonitor的record方法通过logger打印event的message。