序
本文主要研究一下PowerJob的DatabaseMonitorAspect
DatabaseMonitorAspect
tech/powerjob/server/persistence/monitor/DatabaseMonitorAspect.java
代码语言：javascript复制@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
代码语言：javascript复制 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
代码语言：javascript复制public interface MonitorService {
void monitor(Event event);
}MonitorService定义了monitor方法
PowerJobMonitorService
tech/powerjob/server/monitor/PowerJobMonitorService.java
代码语言：javascript复制@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
代码语言：javascript复制public interface Monitor {
/**
* 全局上下文绑定 & 初始化
*/
void init();
/**
* 记录监控事件
* 请注意该方法务必异步不阻塞！！！
* @param event 事件
*/
void record(Event event);
}Monitor接口定义了record方法
LogMonitor
tech/powerjob/server/monitor/monitors/LogMonitor.java
代码语言：javascript复制@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。
