XxlJob是是一个轻量级分布式任务调度平台。对于此可以通过官网进行学习使用。
我们在拥有Sleuth的环境下使用xxjob的时候，就会发现一个问题就是通过xxjob中调用的代码中并不存在链路，当我在XxlJob中出现异常时，需要跨系统查询日志等操作后，得到结论，该操作耗时耗力。所以我们不如向feign，rpc调用一样在xxjob中添加链路。
为了不变更xxjob的源码结构，这里我通过spring aop的方式对@XxlJob注解进行切面编程，而注解@Xxljob就是一个切点。代码如下所示：
将Tracing注入到Aspect切面类中
@EnableAspectJAutoProxy
@Configuration
public class AOPConfiguration {@Beanpublic TraceXXJobAspect traceXXJobAspect(Tracing tracing){return new TraceXXJobAspect(tracing);}}
TraceXXJobAspect ：
@Aspect
@Slf4j
public class TraceXXJobAspect {private Tracing tracing;public TraceXXJobAspect(Tracing tracing) {this.tracing = tracing;}private static final String CLASS_KEY = "class";private static final String METHOD_KEY = "method";@Around("execution (@com.xxl.job.core.handler.annotation.XxlJob * *.*(..))")public Object traceBackgroundThread(final ProceedingJoinPoint pjp) throws Throwable {Tracer tracer = this.tracing.tracer();//清除scop中原spantracer.withSpanInScope(null);String spanName = SpanNameUtil.toLowerHyphen(pjp.getSignature().getName());Span span = tracer.newTrace().name(spanName);try (Tracer.SpanInScope ws = tracer.withSpanInScope(span.start())) {span.tag(CLASS_KEY, pjp.getTarget().getClass().getSimpleName());span.tag(METHOD_KEY, pjp.getSignature().getName());return pjp.proceed();}catch (Throwable ex) {String message = ex.getMessage() == null ? ex.getClass().getSimpleName(): ex.getMessage();span.tag("error", message);throw ex;}finally {span.finish();}}}
这里解释下XxlJob初始化的过程，我们在使用xxljob的时候需要自己创建XxlJobSpringExecutor类实例对象，并注入到sping中(xxljob依赖于spring)，查看下XxlJobSpringExecutor源码：
public class XxlJobSpringExecutor extends XxlJobExecutor implements ApplicationContextAware, SmartInitializingSingleton, DisposableBean {private static final Logger logger = LoggerFactory.getLogger(XxlJobSpringExecutor.class);// start@Overridepublic void afterSingletonsInstantiated() {// init JobHandler Repository 在2.2之后@JobHandler方式被xxljob给取消了/*initJobHandlerRepository(applicationContext);*/// 扫描xxljob注解并为每一个被xxljob生成 IJobHandler对象initJobHandlerMethodRepository(applicationContext);// refresh GlueFactoryGlueFactory.refreshInstance(1);// super starttry {super.start();} catch (Exception e) {throw new RuntimeException(e);}}// destroy@Overridepublic void destroy() {super.destroy();}/*private void initJobHandlerRepository(ApplicationContext applicationContext) {if (applicationContext == null) {return;}// init job handler actionMap<String, Object> serviceBeanMap = applicationContext.getBeansWithAnnotation(JobHandler.class);if (serviceBeanMap != null && serviceBeanMap.size() > 0) {for (Object serviceBean : serviceBeanMap.values()) {if (serviceBean instanceof IJobHandler) {String name = serviceBean.getClass().getAnnotation(JobHandler.class).value();IJobHandler handler = (IJobHandler) serviceBean;if (loadJobHandler(name) != null) {throw new RuntimeException("xxl-job jobhandler[" + name + "] naming conflicts.");}registJobHandler(name, handler);}}}}*/private void initJobHandlerMethodRepository(ApplicationContext applicationContext) {if (applicationContext == null) {return;}// init job handler from method//这里是获取springioc容器的 bean对象String[] beanDefinitionNames = applicationContext.getBeanNamesForType(Object.class, false, true);for (String beanDefinitionName : beanDefinitionNames) {//获取beanObject bean = applicationContext.getBean(beanDefinitionName);Map<Method, XxlJob> annotatedMethods = null; // referred to ：org.springframework.context.event.EventListenerMethodProcessor.processBeantry {//从Bean中判断是否有方法使用了@Xxljob注解annotatedMethods = MethodIntrospector.selectMethods(bean.getClass(),new MethodIntrospector.MetadataLookup<XxlJob>() {@Overridepublic XxlJob inspect(Method method) {return AnnotatedElementUtils.findMergedAnnotation(method, XxlJob.class);}});} catch (Throwable ex) {logger.error("xxl-job method-jobhandler resolve error for bean[" + beanDefinitionName + "].", ex);}//没有获取到开始下个循环if (annotatedMethods==null || annotatedMethods.isEmpty()) {continue;}for (Map.Entry<Method, XxlJob> methodXxlJobEntry : annotatedMethods.entrySet()) {//获取xxljob对应的method对象Method method = methodXxlJobEntry.getKey();//获取xxljob注解对象XxlJob xxlJob = methodXxlJobEntry.getValue();if (xxlJob == null) {continue;}//对获取nameString name = xxlJob.value();if (name.trim().length() == 0) {throw new RuntimeException("xxl-job method-jobhandler name invalid, for[" + bean.getClass() + "#" + method.getName() + "] .");}//验证name的唯一性 if (loadJobHandler(name) != null) {throw new RuntimeException("xxl-job jobhandler[" + name + "] naming conflicts.");}// execute method//日志记录if (!(method.getParameterTypes().length == 1 && method.getParameterTypes()[0].isAssignableFrom(String.class))) {throw new RuntimeException("xxl-job method-jobhandler param-classtype invalid, for[" + bean.getClass() + "#" + method.getName() + "] , " +"The correct method format like \" public ReturnT<String> execute(String param) \" .");}if (!method.getReturnType().isAssignableFrom(ReturnT.class)) {throw new RuntimeException("xxl-job method-jobhandler return-classtype invalid, for[" + bean.getClass() + "#" + method.getName() + "] , " +"The correct method format like \" public ReturnT<String> execute(String param) \" .");}method.setAccessible(true);// init and destoryMethod initMethod = null;Method destroyMethod = null;//根据init属性判断是否存在对应init方法if (xxlJob.init().trim().length() > 0) {try {initMethod = bean.getClass().getDeclaredMethod(xxlJob.init());initMethod.setAccessible(true);} catch (NoSuchMethodException e) {throw new RuntimeException("xxl-job method-jobhandler initMethod invalid, for[" + bean.getClass() + "#" + method.getName() + "] .");}}//根据init属性判断是否存在对应destroy方法if (xxlJob.destroy().trim().length() > 0) {try {destroyMethod = bean.getClass().getDeclaredMethod(xxlJob.destroy());destroyMethod.setAccessible(true);} catch (NoSuchMethodException e) {throw new RuntimeException("xxl-job method-jobhandler destroyMethod invalid, for[" + bean.getClass() + "#" + method.getName() + "] .");}}// registry jobhandlerregistJobHandler(name, new MethodJobHandler(bean, method, initMethod, destroyMethod));}}}// ---------------------- applicationContext ----------------------private static ApplicationContext applicationContext;@Overridepublic void setApplicationContext(ApplicationContext applicationContext) throws BeansException {this.applicationContext = applicationContext;}public static ApplicationContext getApplicationContext() {return applicationContext;}}
注册jobhandler，实际上写入到本地缓存中jobHandlerRepository，如下所示：
// ---------------------- job handler repository ----------------------private static ConcurrentMap<String, IJobHandler> jobHandlerRepository = new ConcurrentHashMap<String, IJobHandler>();public static IJobHandler registJobHandler(String name, IJobHandler jobHandler){logger.info(">>>>>>>>>>> xxl-job register jobhandler success, name:{}, jobHandler:{}", name, jobHandler);return jobHandlerRepository.put(name, jobHandler);}
xxljob的调用时基于Netty的，客户端在初始化的时候会本地构建一个Netty服务端(详细可参考NettyHttpServer)，并与admin保持一个长连接，在Admin来被定时调度的触发后通过Netty发送调度的信息给客户端(详细参考NettyServerHandler与XxlRpcProviderFactory)，客户端收到信息后，拿着信息去jobHandlerRepository中找到对应的IjobHandler对象，执行对应的方法