前往小程序,Get更优阅读体验!
立即前往
首页
学习
活动
专区
工具
TVP
发布
社区首页 >专栏 >聊聊PowerJob的ProcessorLoader

聊聊PowerJob的ProcessorLoader

作者头像
code4it
发布2023-12-26 18:02:47
1500
发布2023-12-26 18:02:47
举报
文章被收录于专栏:码匠的流水账码匠的流水账

本文主要研究一下PowerJob的ProcessorLoader

ProcessorLoader

tech/powerjob/worker/processor/ProcessorLoader.java

代码语言:javascript
复制
public interface ProcessorLoader {

    ProcessorBean load(ProcessorDefinition definition);
}

ProcessorLoader定义了load方法,用于根据ProcessorDefinition加载ProcessorBean

ProcessorDefinition

tech/powerjob/worker/extension/processor/ProcessorDefinition.java

代码语言:javascript
复制
@Getter
@Setter
@ToString
@Accessors(chain = true)
public class ProcessorDefinition implements Serializable {

    /**
     * 后台配置的处理器类型
     */
    private String processorType;
    /**
     * 后台配置的处理器信息
     */
    private String processorInfo;

    @Override
    public boolean equals(Object o) {
        if (this == o) {
            return true;
        }
        if (o == null || getClass() != o.getClass()) {
            return false;
        }
        ProcessorDefinition that = (ProcessorDefinition) o;
        return Objects.equals(processorType, that.processorType) && Objects.equals(processorInfo, that.processorInfo);
    }

    @Override
    public int hashCode() {
        return Objects.hash(processorType, processorInfo);
    }
}

ProcessorDefinition定义了processorType及processorInfo两个属性

ProcessorBean

tech/powerjob/worker/extension/processor/ProcessorBean.java

代码语言:javascript
复制
@Getter
@Setter
@Accessors(chain = true)
public class ProcessorBean {

    /**
     * 真正用来执行逻辑的处理器对象
     */
    private transient BasicProcessor processor;
    /**
     * 加载该处理器对象的 classLoader,可空,空则使用 {@link Object#getClass()#getClassLoader() 代替}
     */
    private transient ClassLoader classLoader;

}

ProcessorBean定义了BasicProcessor及ClassLoader两个属性

PowerJobProcessorLoader

tech/powerjob/worker/processor/PowerJobProcessorLoader.java

代码语言:javascript
复制
@Slf4j
public class PowerJobProcessorLoader implements ProcessorLoader {

    private final List<ProcessorFactory> processorFactoryList;
    private final Map<ProcessorDefinition, ProcessorBean> def2Bean = new ConcurrentHashMap<>(128);

    public PowerJobProcessorLoader(List<ProcessorFactory> processorFactoryList) {
        this.processorFactoryList = processorFactoryList;
    }

    @Override
    public ProcessorBean load(ProcessorDefinition definition) {
        return def2Bean.computeIfAbsent(definition, ignore -> {
            final String processorType = definition.getProcessorType();
            log.info("[ProcessorFactory] start to load Processor: {}", definition);
            for (ProcessorFactory pf : processorFactoryList) {
                final String pfName = pf.getClass().getSimpleName();
                if (!Optional.ofNullable(pf.supportTypes()).orElse(Collections.emptySet()).contains(processorType)) {
                    log.info("[ProcessorFactory] [{}] can't load type={}, skip!", pfName, processorType);
                    continue;
                }
                log.info("[ProcessorFactory] [{}] try to load processor: {}", pfName, definition);
                try {
                    ProcessorBean processorBean = pf.build(definition);
                    if (processorBean != null) {
                        log.info("[ProcessorFactory] [{}] load processor successfully: {}", pfName, definition);
                        return processorBean;
                    }
                } catch (Throwable t) {
                    log.error("[ProcessorFactory] [{}] load processor failed: {}", pfName, definition, t);
                }
            }
            throw new PowerJobException("fetch Processor failed, please check your processorType and processorInfo config");
        });
    }
}

PowerJobProcessorLoader实现了ProcessorLoader接口,其构造器要求传入processorFactoryList,它还定义了def2Bean,用于维护ProcessorDefinition与ProcessorBean的关系;其load方法使用ConcurrentHashMap的computeIfAbsent,将加载好的ProcessorBean放入到def2Bean;其加载过程为遍历processorFactoryList,找到支持该processorType的ProcessorFactory,然后执行其build方法进行构造

ProcessorFactory

tech/powerjob/worker/extension/processor/ProcessorFactory.java

代码语言:javascript
复制
public interface ProcessorFactory {

    /**
     * 支持的处理器类型,类型不匹配则跳过该 ProcessorFactory 的加载逻辑
     * 对应的是控制台的'处理器类型' TAB,不做任何定制的情况下,取值范围为 {@link ProcessorType#name()}
     * @return 支持的处理器类型
     */
    Set<String> supportTypes();

    /**
     * 根据处理器定义构建处理器对象
     * 注意:Processor 为单例对象,即 PowerJob 对每一个 ProcessorBean 只调用一次 build 方法
     * @param processorDefinition 处理器定义
     * @return null or ProcessorBean
     */
    ProcessorBean build(ProcessorDefinition processorDefinition);
}

ProcessorFactory接口定义了supportTypes、build方法;它有四个实现类,其中BuiltInSpringProcessorFactory及BuildInSpringMethodProcessorFactory继承自AbstractBuildInSpringProcessorFactory,另外两个为BuiltInDefaultProcessorFactory、JarContainerProcessorFactory

BuiltInDefaultProcessorFactory

tech/powerjob/worker/processor/impl/BuiltInDefaultProcessorFactory.java

代码语言:javascript
复制
@Slf4j
public class BuiltInDefaultProcessorFactory implements ProcessorFactory {

    @Override
    public Set<String> supportTypes() {
        return Sets.newHashSet(ProcessorType.BUILT_IN.name());
    }

    @Override
    public ProcessorBean build(ProcessorDefinition processorDefinition) {

        String className = processorDefinition.getProcessorInfo();

        try {
            Class<?> clz = Class.forName(className);
            BasicProcessor basicProcessor = (BasicProcessor) clz.getDeclaredConstructor().newInstance();
            return new ProcessorBean()
                    .setProcessor(basicProcessor)
                    .setClassLoader(basicProcessor.getClass().getClassLoader());
        }catch (Exception e) {
            log.warn("[ProcessorFactory] load local Processor(className = {}) failed.", className, e);
        }
        return null;
    }
}

BuiltInDefaultProcessorFactory是默认的处理器工厂,通过全限定类名加载处理器,但没有IOC功能

JarContainerProcessorFactory

tech/powerjob/worker/processor/impl/JarContainerProcessorFactory.java

代码语言:javascript
复制
@Slf4j
public class JarContainerProcessorFactory implements ProcessorFactory {

    private final WorkerRuntime workerRuntime;

    public JarContainerProcessorFactory(WorkerRuntime workerRuntime) {
        this.workerRuntime = workerRuntime;
    }

    @Override
    public Set<String> supportTypes() {
        return Sets.newHashSet(ProcessorType.EXTERNAL.name());
    }

    @Override
    public ProcessorBean build(ProcessorDefinition processorDefinition) {

        String processorInfo = processorDefinition.getProcessorInfo();
        String[] split = processorInfo.split("#");
        String containerName = split[0];
        String className = split[1];

        log.info("[ProcessorFactory] try to load processor({}) in container({})", className, containerName);

        OmsContainer omsContainer = OmsContainerFactory.fetchContainer(Long.valueOf(containerName), workerRuntime);
        if (omsContainer != null) {
            return new ProcessorBean()
                    .setProcessor(omsContainer.getProcessor(className))
                    .setClassLoader(omsContainer.getContainerClassLoader());
        } else {
            log.warn("[ProcessorFactory] load container failed. processor info : {}", processorInfo);
        }
        return null;
    }
}

JarContainerProcessorFactory主要是通过OmsContainer来加载ProcessorBean

AbstractBuildInSpringProcessorFactory

tech/powerjob/worker/processor/impl/AbstractBuildInSpringProcessorFactory.java

代码语言:javascript
复制
@Slf4j
public abstract class AbstractBuildInSpringProcessorFactory implements ProcessorFactory {

    protected final ApplicationContext applicationContext;

    protected AbstractBuildInSpringProcessorFactory(ApplicationContext applicationContext) {
        this.applicationContext = applicationContext;
    }

    @Override
    public Set<String> supportTypes() {
        return Sets.newHashSet(ProcessorType.BUILT_IN.name());
    }

    protected boolean checkCanLoad() {
        try {
            ApplicationContext.class.getClassLoader();
            return applicationContext != null;
        } catch (Throwable ignore) {
        }
        return false;
    }


    @SuppressWarnings("unchecked")
    protected static <T> T getBean(String className, ApplicationContext ctx) throws Exception {

        // 0. 尝试直接用 Bean 名称加载
        try {
            final Object bean = ctx.getBean(className);
            if (bean != null) {
                return (T) bean;
            }
        } catch (Exception ignore) {
        }

        // 1. ClassLoader 存在,则直接使用 clz 加载
        ClassLoader classLoader = ctx.getClassLoader();
        if (classLoader != null) {
            return (T) ctx.getBean(classLoader.loadClass(className));
        }
        // 2. ClassLoader 不存在(系统类加载器不可见),尝试用类名称小写加载
        String[] split = className.split("\\.");
        String beanName = split[split.length - 1];
        // 小写转大写
        char[] cs = beanName.toCharArray();
        cs[0] += 32;
        String beanName0 = String.valueOf(cs);
        log.warn("[SpringUtils] can't get ClassLoader from context[{}], try to load by beanName:{}", ctx, beanName0);
        return (T) ctx.getBean(beanName0);
    }
}

AbstractBuildInSpringProcessorFactory是两个spring相关ProcessorFactory的抽象类,它使用spring的ApplicationContext来加载

BuiltInSpringProcessorFactory

tech/powerjob/worker/processor/impl/BuiltInSpringProcessorFactory.java

代码语言:javascript
复制
@Slf4j
public class BuiltInSpringProcessorFactory extends AbstractBuildInSpringProcessorFactory {


    public BuiltInSpringProcessorFactory(ApplicationContext applicationContext) {
        super(applicationContext);
    }

    @Override
    public ProcessorBean build(ProcessorDefinition processorDefinition) {

        try {
            boolean canLoad = checkCanLoad();
            if (!canLoad) {
                log.info("[ProcessorFactory] can't find Spring env, this processor can't load by 'BuiltInSpringProcessorFactory'");
                return null;
            }
            String processorInfo = processorDefinition.getProcessorInfo();
            //用于区分方法级别的参数
            if (processorInfo.contains("#")) {
                return null;
            }
            BasicProcessor basicProcessor = getBean(processorInfo, applicationContext);
            return new ProcessorBean()
                    .setProcessor(basicProcessor)
                    .setClassLoader(basicProcessor.getClass().getClassLoader());
        } catch (NoSuchBeanDefinitionException ignore) {
            log.warn("[ProcessorFactory] can't find the processor in SPRING");
        } catch (Throwable t) {
            log.warn("[ProcessorFactory] load by BuiltInSpringProcessorFactory failed. If you are using Spring, make sure this bean was managed by Spring", t);
        }

        return null;
    }
}

BuiltInSpringProcessorFactory通过ApplicationContext加载spring相关的Bean,但它不处理processorInfo包含#的processorDefinition

BuildInSpringMethodProcessorFactory

tech/powerjob/worker/processor/impl/BuildInSpringMethodProcessorFactory.java

代码语言:javascript
复制
@Slf4j
public class BuildInSpringMethodProcessorFactory extends AbstractBuildInSpringProcessorFactory {

    private static final List<String> jobHandlerRepository = new LinkedList<>();

    private final static String DELIMITER = "#";


    public BuildInSpringMethodProcessorFactory(ApplicationContext applicationContext) {
        super(applicationContext);
    }


    @Override
    public ProcessorBean build(ProcessorDefinition processorDefinition) {
        try {
            boolean canLoad = checkCanLoad();
            if (!canLoad) {
                log.info("[ProcessorFactory] can't find Spring env, this processor can't load by 'BuildInSpringMethodProcessorFactory'");
                return null;
            }
            String processorInfo = processorDefinition.getProcessorInfo();
            if (!processorInfo.contains(DELIMITER)) {
                log.info("[ProcessorFactory] can't parse processorDefinition, this processor can't load by 'BuildInSpringMethodProcessorFactory'");
                return null;
            }
            String[] split = processorInfo.split(DELIMITER);
            String methodName = split[1];
            String className = split[0];
            Object bean = getBean(className,applicationContext);
            Method[] methods = bean.getClass().getDeclaredMethods();
            for (Method method : methods) {
                PowerJobHandler powerJob = method.getAnnotation(PowerJobHandler.class);
                if (powerJob == null) {
                    continue;
                }
                String name = powerJob.name();
                //匹配到和页面定义相同的methodName
                if (!name.equals(methodName)) {
                    continue;
                }
                if (name.trim().length() == 0) {
                    throw new RuntimeException("method-jobhandler name invalid, for[" + bean.getClass() + "#" + method.getName() + "] .");
                }
                if (containsJobHandler(name)) {
                    throw new RuntimeException("jobhandler[" + name + "] naming conflicts.");
                }
                method.setAccessible(true);
                registerJobHandler(methodName);

                MethodBasicProcessor processor = new MethodBasicProcessor(bean, method);
                return new ProcessorBean()
                        .setProcessor(processor)
                        .setClassLoader(processor.getClass().getClassLoader());
            }
        } catch (NoSuchBeanDefinitionException ignore) {
            log.warn("[ProcessorFactory] can't find the processor in SPRING");
        } catch (Throwable t) {
            log.warn("[ProcessorFactory] load by BuiltInSpringProcessorFactory failed. If you are using Spring, make sure this bean was managed by Spring", t);
        }
        return null;
    }


    public static void registerJobHandler(String name) {
        jobHandlerRepository.add(name);
    }


    private boolean containsJobHandler(String name) {
        return jobHandlerRepository.contains(name);
    }
}

BuildInSpringMethodProcessorFactory专门用于processorInfo包含#的processorDefinition,它会遍历指定class的methods,找到方法上标注有@PowerJobHandler注解且方法名一致的method,注册到jobHandlerRepository,其创建的是MethodBasicProcessor

小结

PowerJob的ProcessorLoader定义了load方法,用于根据ProcessorDefinition加载ProcessorBean;PowerJobProcessorLoader实现了ProcessorLoader接口,它会遍历processorFactoryList,找到支持该processorType的ProcessorFactory,然后执行其build方法进行构造;ProcessorFactory接口定义了supportTypes、build方法;它有四个实现类,其中BuiltInSpringProcessorFactory及BuildInSpringMethodProcessorFactory继承自AbstractBuildInSpringProcessorFactory,另外两个为BuiltInDefaultProcessorFactory、JarContainerProcessorFactory。

本文参与 腾讯云自媒体同步曝光计划,分享自微信公众号。
原始发表:2023-12-22,如有侵权请联系 cloudcommunity@tencent.com 删除

本文分享自 码匠的流水账 微信公众号,前往查看

如有侵权,请联系 cloudcommunity@tencent.com 删除。

本文参与 腾讯云自媒体同步曝光计划  ,欢迎热爱写作的你一起参与!

评论
登录后参与评论
0 条评论
热度
最新
推荐阅读
目录
  • ProcessorLoader
    • ProcessorDefinition
      • ProcessorBean
      • PowerJobProcessorLoader
      • ProcessorFactory
        • BuiltInDefaultProcessorFactory
          • JarContainerProcessorFactory
            • AbstractBuildInSpringProcessorFactory
              • BuiltInSpringProcessorFactory
              • BuildInSpringMethodProcessorFactory
          • 小结
          领券
          问题归档专栏文章快讯文章归档关键词归档开发者手册归档开发者手册 Section 归档