Play源码深入之五:Job模块的原理
Play源码深入之五:Job模块的原理
刀狂剑痴 发表于2年前
Play源码深入之五:Job模块的原理
  • 发表于 2年前
  • 阅读 808
  • 收藏 0
  • 点赞 0
  • 评论 0

腾讯云 新注册用户 域名抢购1元起>>>   

先看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
博文 111
码字总数 82582
×
刀狂剑痴
如果觉得我的文章对您有用,请随意打赏。您的支持将鼓励我继续创作!
* 金额(元)
¥1 ¥5 ¥10 ¥20 其他金额
打赏人
留言
* 支付类型
微信扫码支付
打赏金额:
已支付成功
打赏金额: