文档章节

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

奋斗到天明
 奋斗到天明
发布于 2015/08/27 18:12
字数 1007
阅读 827
收藏 0
点赞 0
评论 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 ⋅ 0

《Apache Spark 设计与实现》

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

叶秀兰 ⋅ 2015/07/16 ⋅ 1

sharding-jdbc源码分析—准备工作

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

飞哥-Javaer ⋅ 05/03 ⋅ 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

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

基于spring+quartz的分布式任务调度

学习地址:http://www.roncoo.com/course/view/e2b459016e2e477dbd5d67c8b23fe86d 课程介绍 Quartz是OpenSymphony开源组织在Job scheduling领域又一个开源项目,它可以与J2EE与J2SE应用程序相...

小红牛 ⋅ 04/19 ⋅ 0

Play源码深入之一:从play命令开始

Play的命令是借助python脚本,这从下载的play包就能明显看出来:一是其中有一个python包,里面是一个play自带的python环境,还有是play.bat文件: @echo off"%~dp0pythonpython.exe" "%~dp0p...

刀狂剑痴 ⋅ 2015/08/27 ⋅ 0

2018年的学习计划

转眼间,2017年过去了,在过去的一年里,成长颇多,感受颇多,当然,收获也颇多。能够在2017年里,将quartz,xxl-job,elastic-job-Lite源码读一遍,学到了很多东西。 整理了一下自己感兴趣的...

一滴水的坚持 ⋅ 01/06 ⋅ 0

源码圈 365 胖友的书单整理

🙂🙂🙂关注微信公众号:【芋道源码】有福利: RocketMQ / MyCAT / Sharding-JDBC 所有源码分析文章列表 RocketMQ / MyCAT / Sharding-JDBC 中文注释源码 GitHub 地址 您对于源码的疑问...

芋道源码掘金Java群217878901 ⋅ 2017/09/21 ⋅ 0

Android动态化框架App Bundles

Android App Bundles 在今年的Google I/O大会上,Google向 Android 引入了新 App 动态化框架(即Android App Bundle,缩写为AAB),与Instant App不同,AAB是借助Split Apk完成动态加载,使用...

code_xzh ⋅ 05/16 ⋅ 0

没有更多内容

加载失败,请刷新页面

加载更多

下一页

使用 vue-cli 搭建项目

vue-cli 是一个官方发布 vue.js 项目脚手架,使用 vue-cli 可以快速创建 vue 项目,GitHub地址是:https://github.com/vuejs/vue-cli 一、 安装 node.js 首先需要安装node环境,可以直接到中...

初学者的优化 ⋅ 12分钟前 ⋅ 0

设计模式 之 享元模式

设计模式 之 享元模式 定义 使用共享技术来有效地支持大量细粒度对象的复用 关键点:防止类多次创建,造成内存溢出; 使用享元模式来将内部状态与外部状态进行分离,在循环创建对象的环境下,...

GMarshal ⋅ 27分钟前 ⋅ 0

SpringBoot集成Druid的最简单的小示例

参考网页 https://blog.csdn.net/king_is_everyone/article/details/53098350 建立maven工程 Pom文件 <?xml version="1.0" encoding="UTF-8"?> <project xmlns="http://maven.apache.org/POM......

karma123 ⋅ 今天 ⋅ 0

Java虚拟机基本结构的简单记忆

Java堆:一般是放置实例化的对象的地方,堆分新生代和老年代空间,不断未被回收的对象越老,被放入老年代空间。分配最大堆空间:-Xmx 分配初始堆空间:-Xms,分配新生代空间:-Xmn,新生代的大小一...

算法之名 ⋅ 今天 ⋅ 0

OSChina 周日乱弹 —— 这么好的姑娘都不要了啊

Osc乱弹歌单(2018)请戳(这里) 【今日歌曲】 @TigaPile :分享曾惜的单曲《讲真的》 《讲真的》- 曾惜 手机党少年们想听歌,请使劲儿戳(这里) @首席搬砖工程师 :怎样约女孩子出来吃饭,...

小小编辑 ⋅ 今天 ⋅ 8

Jenkins实践3 之脚本

#!/bin/sh# export PROJ_PATH=项目路径# export TOMCAT_PATH=tomcat路径killTomcat(){pid=`ps -ef | grep tomcat | grep java|awk '{print $2}'`echo "tom...

晨猫 ⋅ 今天 ⋅ 0

Spring Bean的生命周期

前言 Spring Bean 的生命周期在整个 Spring 中占有很重要的位置,掌握这些可以加深对 Spring 的理解。 首先看下生命周期图: 再谈生命周期之前有一点需要先明确: Spring 只帮我们管理单例模...

素雷 ⋅ 今天 ⋅ 0

zblog2.3版本的asp系统是否可以超越卢松松博客的流量[图]

最近访问zblog官网,发现zlbog-asp2.3版本已经进入测试阶段了,虽然正式版还没有发布,想必也不久了。那么作为aps纵横江湖十多年的今天,blog2.2版本应该已经成熟了,为什么还要发布这个2.3...

原创小博客 ⋅ 今天 ⋅ 0

聊聊spring cloud的HystrixCircuitBreakerConfiguration

序 本文主要研究一下spring cloud的HystrixCircuitBreakerConfiguration HystrixCircuitBreakerConfiguration spring-cloud-netflix-core-2.0.0.RELEASE-sources.jar!/org/springframework/......

go4it ⋅ 今天 ⋅ 0

二分查找

二分查找,也称折半查找、二分搜索,是一种在有序数组中查找某一特定元素的搜索算法。搜素过程从数组的中间元素开始,如果中间元素正好是要查找的元素,则搜素过程结束;如果某一特定元素大于...

人觉非常君 ⋅ 今天 ⋅ 0

没有更多内容

加载失败,请刷新页面

加载更多

下一页

返回顶部
顶部