文档章节

Play源码深入之五:Job模块的原理

奋斗到天明
 奋斗到天明
发布于 2015/08/27 18:12
字数 1007
阅读 873
收藏 0

先看play.jobs.JobsPlugin。

public void onApplicationStart() {
    int core = Integer.parseInt(Play.configuration.getProperty("play.jobs.pool", "10"));
    executor = new ScheduledThreadPoolExecutor(core, new PThreadFactory("jobs"), new ThreadPoolExecutor.AbortPolicy());
}

在onAppliactionStart()方法中,实例化一个ScheduledThreadPollExecutor做executor。 接受afterApplicationStart事件中,才会处理Job。

public void afterApplicationStart() {
    List> jobs = new ArrayList>();
    for (Class clazz : Play.classloader.getAllClasses()) {
        if (Job.class.isAssignableFrom(clazz)) {
            jobs.add(clazz);
        }
    }
    scheduledJobs = new ArrayList();
    for (final Class clazz : jobs) {
        // @OnApplicationStart
        if (clazz.isAnnotationPresent(OnApplicationStart.class)) {
            //check if we're going to run the job sync or async
            OnApplicationStart appStartAnnotation = clazz.getAnnotation(OnApplicationStart.class);
            if( !appStartAnnotation.async()) {
                //run job sync
                try {
                    Job job = ((Job) clazz.newInstance());
                    scheduledJobs.add(job);
                    job.run();
                    if(job.wasError) {
                        if(job.lastException != null) {
                            throw job.lastException;
                        }
                        throw new RuntimeException("@OnApplicationStart Job has failed");
                    }
                } catch (...) {
                    ...
                }
            } else {
                //run job async
                try {
                    Job job = ((Job) clazz.newInstance());
                    scheduledJobs.add(job);
                    //start running job now in the background
                    @SuppressWarnings("unchecked")
                    Callable callable = (Callable)job;
                    executor.submit(callable);
                } catch (...) {
                    ...
                }
            }
        }

        // @On
        if (clazz.isAnnotationPresent(On.class)) {
            try {
                Job job = ((Job) clazz.newInstance());
                scheduledJobs.add(job);
                scheduleForCRON(job);
            } catch (InstantiationException ex) {
                throw new UnexpectedException("Cannot instanciate Job " + clazz.getName());
            } catch (IllegalAccessException ex) {
                throw new UnexpectedException("Cannot instanciate Job " + clazz.getName());
            }
        }
        // @Every
        if (clazz.isAnnotationPresent(Every.class)) {
            try {
                Job job = (Job) clazz.newInstance();
                scheduledJobs.add(job);
                String value = job.getClass().getAnnotation(Every.class).value();
                if (value.startsWith("cron.")) {
                    value = Play.configuration.getProperty(value);
                }
                value = Expression.evaluate(value, value).toString();
                if(!"never".equalsIgnoreCase(value)){
                    executor.scheduleWithFixedDelay(job, Time.parseDuration(value), Time.parseDuration(value), TimeUnit.SECONDS);
                }
            } catch (InstantiationException ex) {
                throw new UnexpectedException("Cannot instanciate Job " + clazz.getName());
            } catch (IllegalAccessException ex) {
                throw new UnexpectedException("Cannot instanciate Job " + clazz.getName());
            }   
    }
}

public static  void scheduleForCRON(Job job) {
    ...
    try {
        ...
        executor.schedule((Callable)job, nextDate.getTime() - now.getTime(), TimeUnit.MILLISECONDS);
        ...
    } catch (Exception ex) {
        throw new UnexpectedException(ex);
    }
}

第一步读取所有类中的Job类,并判断是否是OnApplicationStart标记,如果是sync同步的,就会直接run。如果async异步的,会加入executor中,虽然执行时间间隔为0毫秒,但是实际执行时间由executor决定。 如果被On标记,play会解析On注解中表达式的值。

这里需要注意,如果是以cron开头,就会读取配置文件中的值,这点之前还没发现哈。 然后根据表达式触发时间与目前时间计算延迟时间,并加入executor中。 

如果是Every标记,也会像上面的一样处理注解中表达式,不同的是,play会将这个Job和他的执行时间注入到executor,不用再手动规定延迟执行时间,由executor完全接管执行。 

那么被On标记的job如何达到周期执行的效果呢?关键在play.jobs.Job类中。

public class Job extends Invoker.Invocation implements Callable {

    public void doJob() throws Exception {
    }

    public V doJobWithResult() throws Exception {
        doJob();
        return null;
    }

    private V withinFilter(play.libs.F.Function0 fct) throws Throwable {
        for (PlayPlugin plugin :  Play.pluginCollection.getEnabledPlugins() ){
           if (plugin.getFilter() != null) {
              return (V)plugin.getFilter().withinFilter(fct);
           }
        }
        return null;
    }

    public V call() {
        Monitor monitor = null;
        try {
            if (init()) {
                before();
                V result = null;

                try {
                    ... 
                    result = withinFilter(new play.libs.F.Function0() {
                        public V apply() throws Throwable {
                          return doJobWithResult();
                        }
                    });
                    ...
                } catch (...) {
                    ...
                }
                after();
                return result;
            }
        } catch (Throwable e) {
            onException(e);
        } finally {
            ...
            _finally();
        }
        return null;
    }

    @Override
    public void _finally() {
        super._finally();
        if (executor == JobsPlugin.executor) {
            JobsPlugin.scheduleForCRON(this);
        }
    }
}

run()方法中调用call方法。

call方法中的代码也有init/before/after与请求调用过程类似,因为Job也得初始化相应的上下文。也要通过过滤器,这样JPA事务管理也对Job有用,过滤之后会执行doJobWithReslut()方法,这样就出了框架,进入应用了。

在finally模块中进入__finally()方法后,判断job.executor与JobsPlugin.executor的是否一样,因为在上面处理之中,用On标记的赋值了executor,而every标记的没有赋值。这就是区别,相同的话会再次进入JobsPlugin.scheduleForCRON方法,根据表达式触发时间与目前时间计算延迟时间,并加入executor中。 

如此一次一次就达到周期执行效果。 Job中其他几种算法:

public void every(String delay) {
        every(Time.parseDuration(delay));
    }

    public void every(int seconds) {
        JobsPlugin.executor.scheduleWithFixedDelay(this, seconds, seconds, TimeUnit.SECONDS);
    }

every方法,可以达到设定every注解一样的效果,我猜测应该是匿名内部类时使用,因为不能直接加注解,可以调用every方法。

public Promise in(String delay) {
        return in(Time.parseDuration(delay));
    }
  
    public Promise in(int seconds) {
        final Promise smartFuture = new Promise();
        JobsPlugin.executor.schedule(getJobCallingCallable(smartFuture), seconds, TimeUnit.SECONDS);
        return smartFuture;
    }

in方法,设定延迟时间,延迟执行job。

public Promise now() {
        final Promise smartFuture = new Promise();
        JobsPlugin.executor.submit(getJobCallingCallable(smartFuture));
        return smartFuture;
    }

now方法,立刻执行。

public Promise afterRequest() {
    InvocationContext current = Invoker.InvocationContext.current();
    if(current == null || !Http.invocationType.equals(current.getInvocationType())) {
      return now();
    }

    final Promise smartFuture = new Promise();
    Callable callable = getJobCallingCallable(smartFuture);
    JobsPlugin.addAfterRequestAction(callable);
    return smartFuture;
  }

afterRequest方法,请求执行,利用事件机制,在beforeInvocation与afterInvocation事件清除和执行job。 

in/now/afterRequest三个方法,返回的是Promise对象,是Play对Promise模式的实现,至于什么是Premise模式,大家可以google一下。

© 著作权归作者所有

共有 人打赏支持
奋斗到天明
粉丝 18
博文 112
码字总数 82707
作品 0
昌平
程序员
私信 提问
Play源码解析计划

最近有想法看看Play的源码,以提高自己的编码水平。之前都是东看看,西看看。最后看来去却好像无所大成。有人说过,伤敌十指,不如断敌一指,于是我有开始了学习之路。 原计划是采用1.2.3版本...

刀狂剑痴
2015/08/27
143
0
《Apache Spark 设计与实现》

本文主要讨论 Apache Spark 的设计与实现,重点关注其设计思想、运行原理、实现架构及性能调优,附带讨论与 HadoopMapReduce在设计与实现上的区别。不喜欢将该文档称之为“源码分析”,因为本...

叶秀兰
2015/07/16
8.3K
1
sharding-jdbc源码分析—准备工作

原文作者:阿飞Javaer 原文链接:https://www.jianshu.com/p/7831817c1da8 接下来对sharding-jdbc源码的分析基于tag为源码,根据sharding-jdbc Features深入学习sharding-jdbc的几个主要特性...

飞哥-Javaer
05/03
0
0
Play框架拾遗之五:其他知识点

1、Job异步处理 Job可以有结果返回: package jobs;import play.jobs.; public class MyJob extends Job<String> { public String doJobWithResult() { // 执行一些业务逻辑 return result; }......

奋斗到天明
2015/08/27
0
0
Google I/O 之 Android App Bundles 是个啥

  Android App Bundles(以下简称AAB)是今年Google I/O大会带来的一款全新动态化框架,与Instant App不同,AAB是借助Split Apk完成动态加载。介绍AAB之前,先来了解下SplitApk。   Spl...

Android群英传
05/15
0
0

没有更多内容

加载失败,请刷新页面

加载更多

自定义 Maven 的 repositories

有时,应用中需要一些比较新的依赖,而这些依赖并没有正式发布,还是处于milestone或者是snapshot阶段,并不能从中央仓库或者镜像站上下载到。此时,就需要 自定义Maven的<repositories>。 ...

waylau
22分钟前
1
0
徒手写一个es6代码库

mkdir democd demonpm initnpm install -g babelnpm install -g babel-clinpm install --save-dev babel-preset-es2015-node5 在项目目录创建两个文件夹 functional-playground ......

lilugirl
23分钟前
1
0
linux定位应用问题的一些常用命令,特别针对内存和线程分析的dump命令

1.jps找出进程号,找到对应的进程号后面才好继续操作 2.linux查看进程详细信息 ps -ef | grep 进程ID 3. dump内存信息 Jmap -dump:format=b,file=YYMMddhhmm.dump pid 4.top查看cpu占用信息 ...

noob_chr
23分钟前
1
0
Android TV开发-按键焦点

写在前面 按键焦点过程了解 2.1 dispatchKeyEvent 过程了解 2.2 焦点查找请求过程了解 1.2.1 第一次获取焦点 1.2.3 按键焦点 焦点控制 焦点记忆 应用场景 参考资料 [TOC] 1. 写在前面 工...

冰雪情缘l
23分钟前
1
0
java框架学习日志-3

这章主要是补充一些ioc创建对象的方式,ioc容器在写好<bean></bean>的时候就已经创建对象了。在之前的例子中,一直都是无参的构造方法。下面给出有参的构造方法的对象的创建,没有什么难点重...

白话
26分钟前
2
0

没有更多内容

加载失败,请刷新页面

加载更多

返回顶部
顶部