Java 类java.util.concurrent.RunnableScheduledFuture 实例源码

/**
 * Called in order to initialize a connection with a remote port...
 *
 * @param host
 * @param port
 */
public String initTransfer(final String host, final int port, final boolean isPullMode,
                           final String[] fileList, final String destDir, final FDTPropsDialog d, final boolean isRecursive) {
    // start by constructing a dummy config
    constructConfig(host, port, isPullMode, fileList, destDir, d, isRecursive);
    HeaderBufferPool.initInstance();
    fdtInternalMonitoringTask = (RunnableScheduledFuture) Utils.getMonitoringExecService().scheduleWithFixedDelay(FDTInternalMonitoringTask.getInstance(), 1, 5, TimeUnit.SECONDS);
    consoleReporting = (RunnableScheduledFuture) Utils.getMonitoringExecService().scheduleWithFixedDelay(ConsoleReportingTask.getInstance(), 1, 2, TimeUnit.SECONDS);
    // the session manager will check the "pull/push" mode and start the FDTSession
    try {
        currentSession = FDTSessionManager.getInstance().addFDTClientSession(port);
        fdtSessionMTask = currentSession.getMonitoringTask();
    } catch (Throwable t) {
        logger.log(Level.WARNING, "Got exception when initiating transfer", t);
        return t.getLocalizedMessage();
    }
    return null;
}

public static void realMain(String... args) throws InterruptedException {
    // our tickle service
    ScheduledExecutorService tickleService =
        new ScheduledThreadPoolExecutor(concurrency) {
            // We override decorateTask() to return a custom
            // RunnableScheduledFuture which explicitly removes
            // itself from the queue after cancellation.
            protected <V> RunnableScheduledFuture<V>
                decorateTask(Runnable runnable,
                             RunnableScheduledFuture<V> task) {
                final ScheduledThreadPoolExecutor exec = this;
                return new CustomRunnableScheduledFuture<V>(task) {
                    // delegate to wrapped task, except for:
                    public boolean cancel(boolean b) {
                        // cancel wrapped task & remove myself from the queue
                        return (task().cancel(b)
                                && exec.remove(this));}};}};

    for (int i = 0; i < concurrency; i++)
        new ScheduledTickle(i, tickleService)
            .setUpdateInterval(25, MILLISECONDS);

    done.await();
    tickleService.shutdown();
    pass();
}

public static void realMain(String... args) throws InterruptedException {
    // our tickle service
    ScheduledExecutorService tickleService =
        new ScheduledThreadPoolExecutor(concurrency) {
            // We override decorateTask() to return a custom
            // RunnableScheduledFuture which explicitly removes
            // itself from the queue after cancellation.
            protected <V> RunnableScheduledFuture<V>
                decorateTask(Runnable runnable,
                             RunnableScheduledFuture<V> task) {
                final ScheduledThreadPoolExecutor exec = this;
                return new CustomRunnableScheduledFuture<V>(task) {
                    // delegate to wrapped task, except for:
                    public boolean cancel(boolean b) {
                        // cancel wrapped task & remove myself from the queue
                        return (task().cancel(b)
                                && exec.remove(this));}};}};

    for (int i = 0; i < concurrency; i++)
        new ScheduledTickle(i, tickleService)
            .setUpdateInterval(25, MILLISECONDS);

    done.await();
    tickleService.shutdown();
    pass();
}

/**
 * Determines whether there are tasks which have started and not completed.
 * 
 * As a side effect, this method removes all tasks which are done but are
 * still in the tracking list.
 * 
 * @return {@code true} is active tasks exist.
 */
public boolean hasActiveTasks() {
    boolean doesHaveTasks = false; 
    synchronized (asyncTasks) {
        if (asyncTasks.isEmpty())
            return false;

        Iterator<RunnableScheduledFuture<?>> i = asyncTasks.iterator();
        while (i.hasNext()) {
            RunnableScheduledFuture<?> task = i.next();
            if (task.isDone())
                 i.remove();
            else
                doesHaveTasks = true;
        }
    }
    return doesHaveTasks;
}

@Override
protected <V> RunnableScheduledFuture<V> decorateTask(
    Runnable runnable, RunnableScheduledFuture<V> r) {
  r = super.decorateTask(runnable, r);
  for (; ; ) {
    final int id = idGenerator.next();

    Task<V> task;

    if (runnable instanceof ProjectRunnable) {
      task = new ProjectTask<>((ProjectRunnable) runnable, r, this, id);
    } else {
      task = new Task<>(runnable, r, this, id);
    }

    if (all.putIfAbsent(task.getTaskId(), task) == null) {
      return task;
    }
  }
}

@Override
protected <V> RunnableScheduledFuture<V> decorateTask(Callable<V> callable, RunnableScheduledFuture<V> task) {
    int priority = PRIORITY_INITIAL;
    if (task.getDelay(DEFAULT_TIMEUNIT) <= 0) {
        priority += PRIORITY_IMMEDIATE_OFFSET;
    }

    if (callable instanceof UpdateCallable<?>) {
        UpdateCallable<?> updateCallable = (UpdateCallable<?>)callable;
        if (updateCallable.context.getDocument().getDocument() != null) {
            priority += PRIORITY_FOREGROUND_OFFSET;
        }
    }

    return new PriorityInsertionRunnableScheduledFuture<>(task, priority);
}

@Override
protected <V> RunnableScheduledFuture<V> decorateTask(Runnable runnable, RunnableScheduledFuture<V> task) {
    int priority = PRIORITY_INITIAL;
    if (task.getDelay(DEFAULT_TIMEUNIT) <= 0) {
        priority += PRIORITY_IMMEDIATE_OFFSET;
    }

    if (runnable instanceof UpdateCallable<?>) {
        UpdateCallable<?> updateCallable = (UpdateCallable<?>)runnable;
        if (updateCallable.context.getDocument().getDocument() != null) {
            priority += PRIORITY_FOREGROUND_OFFSET;
        }
    }

    return new PriorityInsertionRunnableScheduledFuture<>(task, priority);
}

@Override
protected <V> RunnableScheduledFuture<V> decorateTask(Callable<V> callable, RunnableScheduledFuture<V> task) {
    int priority = PRIORITY_INITIAL;
    if (task.getDelay(DEFAULT_TIMEUNIT) <= 0) {
        priority += PRIORITY_IMMEDIATE_OFFSET;
    }

    if (callable instanceof UpdateCallable<?>) {
        UpdateCallable<?> updateCallable = (UpdateCallable<?>)callable;
        if (updateCallable.context.getDocument().getDocument() != null) {
            priority += PRIORITY_FOREGROUND_OFFSET;
        }
    }

    return new PriorityInsertionRunnableScheduledFuture<>(task, priority);
}

@Override
protected <V> RunnableScheduledFuture<V> decorateTask(Runnable runnable, RunnableScheduledFuture<V> task) {
    int priority = PRIORITY_INITIAL;
    if (task.getDelay(DEFAULT_TIMEUNIT) <= 0) {
        priority += PRIORITY_IMMEDIATE_OFFSET;
    }

    if (runnable instanceof UpdateCallable<?>) {
        UpdateCallable<?> updateCallable = (UpdateCallable<?>)runnable;
        if (updateCallable.context.getDocument().getDocument() != null) {
            priority += PRIORITY_FOREGROUND_OFFSET;
        }
    }

    return new PriorityInsertionRunnableScheduledFuture<>(task, priority);
}

protected <V> RunnableScheduledFuture<V> queue(final RunnableScheduledFuture<V> future) {
    if (isShutdown()) {
        throw new RejectedExecutionException();
    }
    if (getPoolSize() < getCorePoolSize()) {
        prestartCoreThread();
    }

    if (future instanceof DominoFutureTask) {
        DominoFutureTask<?> dft = (DominoFutureTask<?>) future;
        tasks.put(dft.sequenceNumber, dft);
        if (dft.getDelay(TimeUnit.NANOSECONDS) > 0) {
            dft.setState(TaskState.SLEEPING);
        }
    }
    super.getQueue().add(future);
    return future;
}

@Override
protected <V> RunnableScheduledFuture<V> decorateTask(
    Runnable runnable, RunnableScheduledFuture<V> task) {
  // This gets called by ScheduledThreadPoolExecutor before scheduling a Runnable.
  ensureRunning();
  return task;
}

@Override
protected <V> RunnableScheduledFuture<V> decorateTask(
    Callable<V> callable, RunnableScheduledFuture<V> task) {
  // This gets called by ScheduledThreadPoolExecutor before scheduling a Callable.
  ensureRunning();
  return task;
}

/**
 * Method that converts a RunnableScheduledFuture task in a MyScheduledTask task
 */
@Override
protected <V> RunnableScheduledFuture<V> decorateTask(Runnable runnable,
        RunnableScheduledFuture<V> task) {
    MyScheduledTask<V> myTask=new MyScheduledTask<V>(runnable, null, task,this);    
    return myTask;
}

@Override
protected <V> RunnableScheduledFuture<V> decorateTask(Runnable runnable, RunnableScheduledFuture<V> task) {
    RunnableScheduledFuture<V> future = super.decorateTask(runnable, task);
    if (mTasks == null) {
        synchronized (BackgroundScheduledThreadPoolExecutor.class) {
            if (mTasks == null) {
                mTasks = new ConcurrentHashMap<>();
            }
        }
    }
    mTasks.put(future, runnable);
    return future;
}

@Override
protected void afterExecute(Runnable r, Throwable t) {
    super.afterExecute(r, t);
    if (r instanceof RunnableScheduledFuture) {
        RunnableScheduledFuture<?> future = (RunnableScheduledFuture<?>) r;
        Runnable task = mTasks.get(future);
        if (future.isCancelled() || task == null) { // 当Runnable在run里面cancel自己时还会执行afterExecute方法；
            if (LogUtils.isDebug()) {
                LogUtils.d(TAG, "afterExecute.isCancelled.futrue = " + r + ", throwable = " + t);
            }
        } else {
            int futureHashCode = future.hashCode();
            if (LogUtils.isDebug() && mDebugTimes != null) { // LogUtils.isDebug()是动态设置的，有可能在beforeExecute里为false，在afterExecute里为true了；
                LogUtils.d(TAG, "afterExecute.task = " + task
                        + ", time = " + (SystemClock.elapsedRealtime() - mDebugTimes.get(futureHashCode))
                        + ", throwable = " + t
                        + ", sBackgroundExecutor = " + this);
                if (!future.isPeriodic()) {
                    mDebugTimes.remove(futureHashCode);
                }
            }
            if (!future.isPeriodic()) {
                mRunningTasks.remove(task.hashCode());
                mTasks.remove(future);
            }
        }
    }
    checkAndThrowThreadPoolExecutorThrowable(TAG + ".afterExecute", r, t);
}

void doStateMaintenance() {

    while(!isShutdown()) {
        RunnableScheduledFuture<?> toSchedule;
        while((toSchedule = submittedScheduledTasks.poll()) != null)
            delayedTasks.add(toSchedule);
        RunnableScheduledFuture<?> toExecute;
        while((toExecute = delayedTasks.peek()) != null && toExecute.getDelay(TimeUnit.NANOSECONDS) <= 0) {
            delayedTasks.poll();
            immediateExecutor.executeWithoutWakeup(toExecute);
        }

        RunnableScheduledFuture<?> nextTask = delayedTasks.peek();

        // signal current thread as suspended before we actually check work queues.
        // this avoids wakeupWaiter() seeing an inconsistent state
        currentSleeper.set(Thread.currentThread());

        if(executorQueue.isEmpty() && submittedScheduledTasks.isEmpty()) {
            if(nextTask != null)
                LockSupport.parkNanos(nextTask.getDelay(TimeUnit.NANOSECONDS));
            else
                LockSupport.park();
            currentSleeper.set(null);
        } else {
            currentSleeper.set(null);
            // there are unmatched tasks in the queue, return this thread to the pool
            break;
        }
    }


    // reschedule if we fall out of loop
    if(!isShutdown())
        immediateExecutor.executeWithoutWakeup(scheduler);
}

private int cancelAllAsyncTasks(boolean mayInterruptIfRunning) {
    int notCanceled = 0;
    synchronized (asyncTasks) {
        for (RunnableScheduledFuture<?> task : asyncTasks) {
            if (!task.cancel(mayInterruptIfRunning))
                notCanceled++;
        }
        // remove tasks which are done
        hasActiveTasks();
    }
    return notCanceled;
}

private void setTimeout(final long timeoutMillis) {
  if (timeoutMillis > 0) {
    _future = (RunnableScheduledFuture<?>) _dispatchJob.getDispatcher().getJobTimeoutExecutor().schedule(this, timeoutMillis, TimeUnit.MILLISECONDS);
  } else {
    _future = null;
  }
}

void doStateMaintenance() {

    while(!isShutdown()) {
        RunnableScheduledFuture<?> toSchedule;
        while((toSchedule = submittedScheduledTasks.poll()) != null)
            delayedTasks.add(toSchedule);
        RunnableScheduledFuture<?> toExecute;
        while((toExecute = delayedTasks.peek()) != null && toExecute.getDelay(TimeUnit.NANOSECONDS) <= 0) {
            delayedTasks.poll();
            immediateExecutor.executeWithoutWakeup(toExecute);
        }

        RunnableScheduledFuture<?> nextTask = delayedTasks.peek();

        // signal current thread as suspended before we actually check work queues.
        // this avoids wakeupWaiter() seeing an inconsistent state
        currentSleeper.set(Thread.currentThread());

        if(executorQueue.isEmpty() && submittedScheduledTasks.isEmpty()) {
            if(nextTask != null)
                LockSupport.parkNanos(nextTask.getDelay(TimeUnit.NANOSECONDS));
            else
                LockSupport.park();
            currentSleeper.set(null);
        } else {
            currentSleeper.set(null);
            // there are unmatched tasks in the queue, return this thread to the pool
            break;
        }
    }


    // reschedule if we fall out of loop
    if(!isShutdown())
        immediateExecutor.executeWithoutWakeup(scheduler);
}

boolean isAllTasksCancelled() {
    for(RunnableScheduledFuture<?> task: tasks) {
        if (!task.isCancelled()) {
            return false;
        }
    }
    return true;
}

Task(Runnable runnable, RunnableScheduledFuture<V> task, Executor executor, int taskId) {
  this.runnable = runnable;
  this.task = task;
  this.executor = executor;
  this.taskId = taskId;
  this.running = new AtomicBoolean();
  this.startTime = new Date();
}

public MetricsTask(RunnableScheduledFuture<V> delegate) {
  this.delegate = delegate;
  if(isPeriodic()) {
    ((AtomicInteger) gaugeMap.get(KEY_PERIODIC_COUNT)).incrementAndGet();
  } else {
    ((AtomicInteger) gaugeMap.get(KEY_WAITING_COUNT)).incrementAndGet();
  }
}

protected ScheduledExecutorService createExecutor() {
  LOG.info("Creating a new executor for channel [{}]", getId());
  return new ScheduledThreadPoolExecutor(1) {
    @Override
    protected <V> RunnableScheduledFuture<V> decorateTask(Runnable runnable,
                                                          RunnableScheduledFuture<V> task) {
      if (runnable instanceof CancelableRunnable) {
        return new CancelableScheduledFuture<V>((CancelableRunnable) runnable, task);
      }
      return super.decorateTask(runnable, task);
    }
  };
}

@Test
public void test() throws InterruptedException, ExecutionException, TimeoutException {
  CancelableRunnable cancelableRunnable = Mockito.mock(CancelableRunnable.class);
  RunnableScheduledFuture futureTask = Mockito.mock(RunnableScheduledFuture.class);

  CancelableScheduledFuture<Object> future = new CancelableScheduledFuture<Object>(cancelableRunnable, futureTask);
  future.cancel(false);
  Mockito.verify(futureTask, Mockito.times(1)).cancel(false);

  future.run();
  Mockito.verify(futureTask, Mockito.times(1)).run();

  future.cancel(true);
  Mockito.verify(cancelableRunnable, Mockito.times(1)).cancel();
  Mockito.verify(futureTask, Mockito.times(1)).cancel(true);

  future.isDone();
  Mockito.verify(futureTask, Mockito.times(1)).isDone();

  future.isPeriodic();
  Mockito.verify(futureTask, Mockito.times(1)).isPeriodic();

  future.get();
  Mockito.verify(futureTask, Mockito.times(1)).get();

  future.get(100L, TimeUnit.MICROSECONDS);
  Mockito.verify(futureTask, Mockito.times(1)).get(100L, TimeUnit.MICROSECONDS);

  future.getDelay(TimeUnit.HOURS);
  Mockito.verify(futureTask, Mockito.times(1)).getDelay(TimeUnit.HOURS);

  CancelableScheduledFuture<Object> future2 = new CancelableScheduledFuture<Object>(cancelableRunnable, futureTask);

  assertTrue(future.equals(future));
  assertTrue(future.equals(future2));

  assertEquals(future.hashCode(), future.hashCode());
  assertEquals(future.hashCode(), future2.hashCode());

}

protected <V> RunnableScheduledFuture<V> decorateTask(Runnable r, RunnableScheduledFuture<V> task) {
    return new CustomTask<V>(task);
}

protected <V> RunnableScheduledFuture<V> decorateTask(Callable<V> c, RunnableScheduledFuture<V> task) {
    return new CustomTask<V>(task);
}

public CustomRunnableScheduledFuture(RunnableScheduledFuture<V> task) {
    super();
    this.task = task;
}

protected <V> RunnableScheduledFuture<V> decorateTask(
    Runnable r, RunnableScheduledFuture<V> task) {
    decorations.getAndIncrement();
    return task;
}

protected <V> RunnableScheduledFuture<V> decorateTask(
    Callable<V> c, RunnableScheduledFuture<V> task) {
    decorations.getAndIncrement();
    return task;
}

public ListenableScheduledFuture(Runnable runnable, RunnableScheduledFuture<T> future) {
    super(runnable, null);
    this.future = future;
    this.sequenceNumber = sequencer.getAndIncrement();
}

public ListenableScheduledFuture(Callable<T> callable, RunnableScheduledFuture<T> future) {
    super(callable);
    this.future = future;
    this.sequenceNumber = sequencer.getAndIncrement();
}

@Override
protected <V> RunnableScheduledFuture<V> decorateTask(Callable<V> callable, RunnableScheduledFuture<V> task) {
    return new ListenableScheduledFuture<>(callable, task);
}

@Override
protected <V> RunnableScheduledFuture<V> decorateTask(Runnable runnable, RunnableScheduledFuture<V> task) {
    return new ListenableScheduledFuture<>(runnable, task);
}

protected <V> java.util.concurrent.RunnableScheduledFuture<V> decorateTask(
    Runnable runnable, java.util.concurrent.RunnableScheduledFuture<V> task) {
  return new JitteredRunnableScheduledFuture<>(task);
}

protected <V> java.util.concurrent.RunnableScheduledFuture<V> decorateTask(
    Callable<V> callable, java.util.concurrent.RunnableScheduledFuture<V> task) {
  return new JitteredRunnableScheduledFuture<>(task);
}

JitteredRunnableScheduledFuture(RunnableScheduledFuture<V> wrapped) {
  this.wrapped = wrapped;
}

protected <V> RunnableScheduledFuture<V> decorateTask(Runnable r, RunnableScheduledFuture<V> task) {
    return new CustomTask<V>(task);
}

protected <V> RunnableScheduledFuture<V> decorateTask(Callable<V> c, RunnableScheduledFuture<V> task) {
    return new CustomTask<V>(task);
}

public CustomRunnableScheduledFuture(RunnableScheduledFuture<V> task) {
    super();
    this.task = task;
}

protected <V> RunnableScheduledFuture<V> decorateTask(
    Runnable r, RunnableScheduledFuture<V> task) {
    decorations.getAndIncrement();
    return task;
}