深入解析Apache NIFI的调度策略

//创建一个流文件
flowFile = session.create()

//添加一个属性，在FlowFIle中记录一个时间，姑且把这个时间当做本次调度开始时间
flowFile = session.putAttribute(flowFile, 'Time', String.valueOf(System.currentTimeMillis()))

//然后休眠3秒
try {
    Thread.sleep(3000);
    } catch (InterruptedException e) {
        e.printStackTrace();
    }
//将FlowFIle传输到success
session.transfer(flowFile, REL_SUCCESS)

1591270393098
  +5141  大约是5秒
1591270398239
  +5012  大约是5秒
1591270403251

# If a component has no work to do (is "bored"), how long should we wait before checking again for work?
nifi.bored.yield.duration=10 millis

    @Override
    public void onTrigger(final ProcessContext context, final ProcessSessionFactory sessionFactory) {
        //获取ProcessorNode所指向的Processor
        final Processor processor = processorRef.get().getProcessor();
        //Active Task +1
        activateThread();
        try (final NarCloseable narCloseable = NarCloseable.withComponentNarLoader(getExtensionManager(), processor.getClass(), processor.getIdentifier())) {
            //这里调用的是Processor的onTrigger方法逻辑
            processor.onTrigger(context, sessionFactory);
        } finally {
            //Active Task -1
            deactivateThread();
        }
    }

    private final Map<Thread, ActiveTask> activeThreads = new HashMap<>(48);

    private synchronized void activateThread() {
        final Thread thread = Thread.currentThread();
        final Long timestamp = System.currentTimeMillis();
        activeThreads.put(thread, new ActiveTask(timestamp));
    }

// 打印警告日志
log.warn('我被调度了')

//创建一个流文件
flowFile = session.create()

//添加一个属性，在FlowFIle中记录一个时间，姑且把这个时间当做本次调度开始时间
flowFile = session.putAttribute(flowFile, 'Time', String.valueOf(System.currentTimeMillis()))

//然后休眠3秒
try {
    Thread.sleep(3000);
    } catch (InterruptedException e) {
        e.printStackTrace();
    }
//将FlowFIle传输到success
session.transfer(flowFile, REL_SUCCESS)

    private boolean isWorkToDo() {
        //Connectables意思是可以连接的组件，这里判断当前的connectable的所有上游的Connection是否都是来自于它自己
        boolean hasNonLoopConnection = Connectables.hasNonLoopConnection(connectable);
        //对FUNNEL组件特殊处理 这里我们直接跳过不看
        if (connectable.getConnectableType() == ConnectableType.FUNNEL) {
            // Handle Funnel as a special case because it will never be a 'source' component,
            // and also its outgoing connections can not be terminated.
            // Incoming FlowFiles from other components, and at least one outgoing connection are required.
            return connectable.hasIncomingConnection()
                    && hasNonLoopConnection
                    && !connectable.getConnections().isEmpty()
                    && Connectables.flowFilesQueued(connectable);
        }
        // 判断当前的组件是否是Source，也就是我们前面说的源组件。以下任意情况都可以说明此组件是源组件
        // 1 当组件被TriggerWhenEmpty注解标注的时候(这个注解你可以理解为上游Connection没有FLowFIle也执行调度)
        // 2 当组件上游没有Connection的时候(没有输入FlowFIle的Connection)
        // 3 当组件上游的所有Connection都是来自于这个组件自己(自己连自己的一个循环)，没有来自其他组价的Connection
        final boolean isSourceComponent = connectable.isTriggerWhenEmpty()
                // No input connections
                || !connectable.hasIncomingConnection()
                // Every incoming connection loops back to itself, no inputs from other components
                || !hasNonLoopConnection;

        // 如果这个组件是源组件，那么系统认为它是有工作要做的，否则得看上游Connection是否有FlowFIle
        return isSourceComponent || Connectables.flowFilesQueued(connectable);
    }

public final class FlowEngine extends ScheduledThreadPoolExecutor 

//达到给定的延时时间后，执行任务。这里传入的是实现Runnable接口的任务，
//因此通过ScheduledFuture.get()获取结果为null
public ScheduledFuture<?> schedule(Runnable command,
                                       long delay, TimeUnit unit);
//达到给定的延时时间后，执行任务。这里传入的是实现Callable接口的任务，
//因此，返回的是任务的最终计算结果
 public <V> ScheduledFuture<V> schedule(Callable<V> callable,
                                           long delay, TimeUnit unit);

//是以上一个任务开始的时间计时，period时间过去后，
//检测上一个任务是否执行完毕，如果上一个任务执行完毕，
//则当前任务立即执行，如果上一个任务没有执行完毕，则需要等上一个任务执行完毕后立即执行
public ScheduledFuture<?> scheduleAtFixedRate(Runnable command,
                                                  long initialDelay,
                                                  long period,
                                                  TimeUnit unit);
//当达到延时时间initialDelay后，任务开始执行。上一个任务执行结束后到下一次
//任务执行，中间延时时间间隔为delay。以这种方式，周期性执行任务。
public ScheduledFuture<?> scheduleWithFixedDelay(Runnable command,
                                                     long initialDelay,
                                                     long delay,
                                                     TimeUnit unit);

    @Override
    public void doSchedule(final Connectable connectable, final LifecycleState scheduleState) {
        final List<ScheduledFuture<?>> futures = new ArrayList<>();
        final ConnectableTask connectableTask = new ConnectableTask(this, connectable, flowController, contextFactory, scheduleState, encryptor);

        for (int i = 0; i < connectable.getMaxConcurrentTasks(); i++) {
            // 确定要运行的任务并创建它。
            final AtomicReference<ScheduledFuture<?>> futureRef = new AtomicReference<>();

            final Runnable trigger = createTrigger(connectableTask, scheduleState, futureRef);

            // 周期性的去执行这个trigger
            final ScheduledFuture<?> future = flowEngine.scheduleWithFixedDelay(trigger, 0L,
                connectable.getSchedulingPeriod(TimeUnit.NANOSECONDS), TimeUnit.NANOSECONDS);

            // now that we have the future, set the atomic reference so that if the component is yielded we
            // are able to then cancel this future.
            futureRef.set(future);

            // Keep track of the futures so that we can update the ScheduleState.
            futures.add(future);
        }

        scheduleState.setFutures(futures);
        logger.info("Scheduled {} to run with {} threads", connectable, connectable.getMaxConcurrentTasks());
    }    

private Runnable createTrigger(final ConnectableTask connectableTask, final LifecycleState scheduleState, final AtomicReference<ScheduledFuture<?>> futureRef) {
        final Connectable connectable = connectableTask.getConnectable();
        final Runnable yieldDetectionRunnable = new Runnable() {
            @Override
            public void run() {
                // 调用上面说的connectableTask的invoke方法(上面提到过，这里会判断组件是否有工作要做等等逻辑)
                final InvocationResult invocationResult = connectableTask.invoke();
                if (invocationResult.isYield()) {
                    logger.debug("Yielding {} due to {}", connectable, invocationResult.getYieldExplanation());
                }
                // 如果组件yielded, 则取消任务重新提交，然后过了yield时间再运行。
                //(这里说的yield有两种，一种是因为组件异常，一种是因为没有工作可做，我们重点关注第二种)
                final long newYieldExpiration = connectable.getYieldExpiration();
                final long now = System.currentTimeMillis();
                //组件处理数据发生异常(指没有被捕获已处理的那些)时，组件会被惩罚一段时间不再被调度。显然没有工作可做的组件不是这种情况
                if (newYieldExpiration > now) {
                    ...
                } //noWorkYieldNanos就是上面疑问2说的10ms
                else if (noWorkYieldNanos > 0L && invocationResult.isYield()) {
                    // Component itself didn't yield but there was no work to do, so the framework will choose
                    // to yield the component automatically for a short period of time.
                    final ScheduledFuture<?> scheduledFuture = futureRef.get();
                    if (scheduledFuture == null) {
                        return;
                    }
                    if (scheduledFuture.cancel(false)) {
                        synchronized (scheduleState) {
                            if (scheduleState.isScheduled()) {
                                //设置延迟10ms再调度一次
                                final ScheduledFuture<?> newFuture = flowEngine.scheduleWithFixedDelay(this, noWorkYieldNanos,
                                    connectable.getSchedulingPeriod(TimeUnit.NANOSECONDS), TimeUnit.NANOSECONDS);
                                scheduleState.replaceFuture(scheduledFuture, newFuture);
                                futureRef.set(newFuture);
                            }
                        }
                    }
                }
            }
        };

        return yieldDetectionRunnable;
    }

    @GET
    @Consumes(MediaType.WILDCARD)
    @Produces(MediaType.APPLICATION_JSON)
    @Path("activeThreads")
    public int getActiveThreads() {
       return flowController.timerDrivenEngineRef.get().getActiveCount();
    }

public FlowController flowController;

public final AtomicReference<FlowEngine> timerDrivenEngineRef;

if (noWorkYieldNanos > 0L && invocationResult.isYield()) {
                    logger.info("当前组件没有工作可以做，进入当前处理逻辑，防止处理太快我们查不到正在运行的线程，我们睡眠2秒");
                    try {
                        Thread.sleep(2000L);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                    // Component itself didn't yield but there was no work to do, so the framework will choose
                    // to yield the component automatically for a short period of time.
                    ...

@Override
    public synchronized void doSchedule(final Connectable connectable, final LifecycleState scheduleState) {
        final List<AtomicBoolean> existingTriggers = canceledTriggers.get(connectable);
        if (existingTriggers != null) {
            throw new IllegalStateException("Cannot schedule " + connectable + " because it is already scheduled to run");
        }
        //解析CRON表达式
        final String cronSchedule = connectable.getSchedulingPeriod();
        final CronExpression cronExpression;
        try {
            cronExpression = new CronExpression(cronSchedule);
        } catch (final Exception pe) {
            throw new IllegalStateException("Cannot schedule " + connectable + " to run because its scheduling period is not valid");
        }

        final List<AtomicBoolean> triggers = new ArrayList<>();
        // 根据`Concurrent tasks`的配置，循环注册任务
        for (int i = 0; i < connectable.getMaxConcurrentTasks(); i++) {
            final ConnectableTask continuallyRunTask = new ConnectableTask(this, connectable, flowController, contextFactory, scheduleState, encryptor);

            final AtomicBoolean canceled = new AtomicBoolean(false);

            //计算下一次CRON指定的时间点
            final Date initialDate = cronExpression.getTimeAfter(new Date());
            //计算现在的时间点到下一次CRON指定执行的时间点的时间差
            final long initialDelay = initialDate.getTime() - System.currentTimeMillis();

            final Runnable command = new Runnable() {

                private Date nextSchedule = initialDate;
                //达到这个时间点，虚拟机会运行run方法
                @Override
                public void run() {
                    if (canceled.get()) {
                        return;
                    }
                    try {
                        //这里是上面提过的ConnectableTask的invoke方法，如果应该运行，就会调用Processor的onTrigger方法
                        continuallyRunTask.invoke();
                    } catch (final RuntimeException re) {
                        throw re;
                    } catch (final Exception e) {
                        throw new ProcessException(e);
                    }

                    if (canceled.get()) {
                        return;
                    }
                    //本次调度的主要工作做完了之后，再注册下一次延迟执行
                    //计算下一次任务执行的延时时间
                    nextSchedule = getNextSchedule(nextSchedule, cronExpression);
                    final long delay = getDelay(nextSchedule);

                    logger.debug("Finished task for {}; next scheduled time is at {} after a delay of {} milliseconds", connectable, nextSchedule, delay);
                    //注册下一次延时执行的任务
                    flowEngine.schedule(this, delay, TimeUnit.MILLISECONDS);
                }
            };
            //注册首次的延迟执行任务
            flowEngine.schedule(command, initialDelay, TimeUnit.MILLISECONDS);
            triggers.add(canceled);
        }

        canceledTriggers.put(connectable, triggers);
        logger.info("Scheduled {} to run with {} threads on schedule {}", connectable, connectable.getMaxConcurrentTasks(), cronSchedule);
    }

    private static Date getNextSchedule(final Date currentSchedule, final CronExpression cronExpression) {
        // 由于时钟的精度不是毫秒级，因此我们必须检查是否安排了应该在该时间之后运行的下一个时间，否则我们可能最终会运行两次相同的任务
        final Date now = new Date();
        return cronExpression.getTimeAfter(now.after(currentSchedule) ? now : currentSchedule);
    }    

// 打印警告日志
log.warn('我被调度了，本次调度发生时间为'+String.valueOf(System.currentTimeMillis()))

//创建一个流文件
flowFile = session.create()

//添加一个属性，在FlowFIle中记录一个时间，姑且把这个时间当做本次调度开始时间
flowFile = session.putAttribute(flowFile, 'Time', String.valueOf(System.currentTimeMillis()))

//然后休眠10秒
try {
    Thread.sleep(10000);
    } catch (InterruptedException e) {
        e.printStackTrace();
    }
//将FlowFIle传输到success
session.transfer(flowFile, REL_SUCCESS)

1591608805094
    + 15秒
1591608820004
    + 15秒
1591608835002
    + 15秒
1591608850005

final QuartzSchedulingAgent quartzSchedulingAgent = new QuartzSchedulingAgent(this, timerDrivenEngineRef.get(), repositoryContextFactory, encryptor);
final TimerDrivenSchedulingAgent timerDrivenAgent = new TimerDrivenSchedulingAgent(this, timerDrivenEngineRef.get(), repositoryContextFactory, encryptor, this.nifiProperties);