xxl-job分环境注册

叔牙

发布于 2023-06-21 11:07:12

6080

一、背景二、聊一下spring运行环境三、实现方案四、验证五、参考

一、背景

互联网研发环境一般分为开发、测试、灰度(或预发)和线上,开发和测试共用数据库,预发和线上共用数据库,在我们使用分布式调度平台场景中,为了防止开发环境注册和运行调度影响到测试和线上运行,通常的做法是开发和预发布环境不注册调度能力,从而调度平台不会触发相应的调度逻辑。

另外还有一点就是,一般开发都在内网环境,大家都在同一个出口网关,那么注册到调度平台上的ip可能是同一个,反过来调度平台任务触发回调任务执行器的时候无法穿透内网定位到某一位研发人员的开发机地址,无法正常寻址调度。

我们需要做的就是测试和生产环境正常注册和调度,开发和测试环境不注册调度能力。

二、聊一下spring运行环境

spring.profiles.active用来表示spring项目打包运行的环境,容器启动后可以通过接口回调或者bean注入的方式获取。

1.通过ApplicationContextAware获取

@Slf4j
@Component
public class ApplicationContextUtil implements ApplicationContextAware {
    @Override
    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
     Environment environment = applicationContext.getEnvironment();
        String env = environment.getActiveProfiles()[0];
        log.info("current env = {}",env);
    }
}

2.通过EnvironmentAware获取

@Component
@Slf4j
public class EnvironmentTool implements EnvironmentAware {
    @Override
    public void setEnvironment(Environment environment) {
        String env = environment.getActiveProfiles()[0];
        log.info("current env = {}",env);
    }
}

3.通过依赖注入获取

@Autowired
private Environment environment;

或者

@Autowired
private ApplicationContext applicationContext;

4.注入profile属性

@Value("${spring.profiles.active:[]}")
private String[] activeProfiles;

拿到spring.profiles.active后,我们就可以根据既定的环境变量来定义当前运行的环境了,然后基于当前环境做一些个性化的事情。

三、实现方案

1.在自动注入类添加环境注解

正常情况下我们接入xxl-job能力时,需要注入XxlJobExecutor调度执行器,在spring环境用XxlJobSpringExecutor即可。

@ConditionalOnMissingBean(XxlJobSpringExecutor.class)
@Bean
@Role(BeanDefinition.ROLE_INFRASTRUCTURE)
public XxlJobSpringExecutor xxlJobExecutor() throws Exception {
    XxlJobSpringExecutor xxlJobExecutor = new XxlJobSpringExecutor();
    //省略...
    return xxlJobExecutor;
}

基于@Profile注解改造,在暴露执行器的地方加上profile条件,限制测试和生产环境才会激活bean注入。

@ConditionalOnMissingBean(XxlJobSpringExecutor.class)
@Bean
@Role(BeanDefinition.ROLE_INFRASTRUCTURE)
@Profile({"test","prod"})
public XxlJobSpringExecutor xxlJobExecutor() throws Exception {
    XxlJobSpringExecutor xxlJobExecutor = new XxlJobSpringExecutor();
    //省略...
    return xxlJobExecutor;
}

改造后,只有spring.profiles.active等于test或prod时,才会激活XxlJobSpringExecutor的bean注册,而XxlJobSpringExecutor是xxl-job注册和调度的核心能力,在开发和预发布环境XxlJobSpringExecutor不会注册,那么也不会把@XxlJob注解标记的方法注册成执行器到容器中。

2.在调度器类添加环境注解

在定义@XxlJob注解的类上添加@Profile条件注解:

@Component
@Slf4j
@Profile({"test","prod"})
public class xxxJob {
    @XxlJob("xxxJobHandler")
    public ReturnT<String> xxxJobHandler() {
      //省略...
        return ReturnT.SUCCESS;
    }
}

这个为什么能生效,我们需要了解下调度器注册原理,我们使用的是XxlJobSpringExecutor,其实现了ApplicationContextAware和SmartInitializingSingleton接口,在bean实例化之后会调用重写的afterSingletonsInstantiated方法:

@Override
public void afterSingletonsInstantiated() {
    // init JobHandler Repository (for method)
    initJobHandlerMethodRepository(applicationContext);
    // refresh GlueFactory
    GlueFactory.refreshInstance(1);
    // super start
    try {
        super.start();
    } catch (Exception e) {
        throw new RuntimeException(e);
    }
}

然后调用initJobHandlerMethodRepository方法做注册调度执行器的逻辑:

private void initJobHandlerMethodRepository(ApplicationContext applicationContext) {
    // init job handler from method
    String[] beanDefinitionNames = applicationContext.getBeanNamesForType(Object.class, false, true);
    for (String beanDefinitionName : beanDefinitionNames) {
        Object bean = applicationContext.getBean(beanDefinitionName);
        Map<Method, XxlJob> annotatedMethods = null;   // referred to ：org.springframework.context.event.EventListenerMethodProcessor.processBean
        try {
            annotatedMethods = MethodIntrospector.selectMethods(bean.getClass(),
                    new MethodIntrospector.MetadataLookup<XxlJob>() {
                        @Override
                        public 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()) {
            Method executeMethod = methodXxlJobEntry.getKey();
            XxlJob xxlJob = methodXxlJobEntry.getValue();
            // regist
            registJobHandler(xxlJob, bean, executeMethod);
        }
    }
}

该方法逻辑是从applicationContext中遍历bean,获取被@XxlJob标记的方法,然后调用registJobHandler方法注册任务执行器,最后将@XxlJob注解方法包装成MethodJobHandler进行注册:

registJobHandler(name, new MethodJobHandler(bean, executeMethod, initMethod, destroyMethod));

大致流程如下:

通过上述流程我们大致可以了解到,对于使用spring作为基础构件的接入,想要使用xxl-job调度能力,那么一定需要将@XxlJob标记方法所在的类注入到spring容器中,否则不会被XxlJobSpringExecutor扫描到。

所以我们此处正好取了个巧,基于@Profile激活条件,根据既定环境变量来控制@XxlJob方法所在类的bean注册逻辑,从而通过环境变量控制了调度执行器注册逻辑。

3.修改xxl-job调度器注册逻辑

修改XxlJobSpringExecutor源码:

  private void initJobHandlerMethodRepository(ApplicationContext applicationContext) {
      if (applicationContext == null) {
          return;
      }
      String env = applicationContext.getEnvironment().getActiveProfiles()[0];
      if(null != env && (env.startsWith("dev") || env.startsWith("gray"))) {
          logger.warn("initJobHandlerMethodRepository current env is {},no need register xxl-job",env);
          return;
      }
      //省略注册逻辑...
  }

当拿到当前运行环境不是test或者prod时,放弃任务执行器注册。然后需要将xxl-job-core打包到内部私服仓库,pom文件引入的时候替换掉官方的xxl-job-core包即可。

4.修改@XxlJob注解

先修改@XxlJob注解,添加excludeProfiles属性:

@Target({ElementType.METHOD})
@Retention(RetentionPolicy.RUNTIME)
@Inherited
public @interface XxlJob {
    /**
     * jobhandler name
     */
    String value();
    //省略...
    /**
     * 排除的环境
     * @return
     */
    String[] excludeProfiles() default {};
}

然后修改XxlJobSpringExecutor注册调度执行器逻辑:

private void initJobHandlerMethodRepository(ApplicationContext applicationContext) {
    String env = applicationContext.getEnvironment().getActiveProfiles()[0];
    // init job handler from method
    String[] beanDefinitionNames = applicationContext.getBeanNamesForType(Object.class, false, true);
    for (String beanDefinitionName : beanDefinitionNames) {
        // 省略...
        for (Map.Entry<Method, XxlJob> methodXxlJobEntry : annotatedMethods.entrySet()) {
            Method executeMethod = methodXxlJobEntry.getKey();
            XxlJob xxlJob = methodXxlJobEntry.getValue();
            String[] excludeProfiles = xxlJob.excludeProfiles();
            boolean jumperOver = false;
            if(null != excludeProfiles && excludeProfiles.length > 0) {
                for (String excludeProfile : excludeProfiles) {
                    if(Objects.equals(env,excludeProfile)) {
                        jumperOver = true;
                        break;
                    }
                }
            }
            if(jumperOver) {
                continue;
            }
            // regist
            registJobHandler(xxlJob, bean, executeMethod);
        }
    }
}

从解析到的@XxlJob注解中获取excludeProfiles属性,如果包含当前运行环境的spring.profiles.active,那么认为当前环境不应该注册任务执行器,从而实现环境定制化的能力,并且控制的粒度比较细,到单个@XxlJob任务执行器维度。

四、验证

选择其中一种方式验证,@Profile条件注解,其他方式自行学习验证。

修改之前:

@Component
@Slf4j
public class xxxJob {
    @XxlJob("xxxJobHandler")
    public ReturnT<String> xxxJobHandler() {
      //省略...
        return ReturnT.SUCCESS;
    }
}

修改之后:

@Component
@Slf4j
@Profile({"test","prod"})
public class xxxJob {
    @XxlJob("xxxJobHandler")
    public ReturnT<String> xxxJobHandler() {
      //省略...
        return ReturnT.SUCCESS;
    }
}