聊聊ExecutorService的监控（executive.）

序

本文主要研究一下ExecutorService的监控

InstrumentedExecutorService

metrics-core-4.0.2-sources.jar!/com/codahale/metrics/InstrumentedExecutorService.java

/**
 * An {@link ExecutorService} that monitors the number of tasks submitted, running,
 * completed and also keeps a {@link Timer} for the task duration.
 * <p/>
 * It will register the metrics using the given (or auto-generated) name as classifier, e.g:
 * "your-executor-service.submitted", "your-executor-service.running", etc.
 */
public class InstrumentedExecutorService implements ExecutorService {
 private static final AtomicLong NAME_COUNTER = new AtomicLong();
 private final ExecutorService delegate;
 private final Meter submitted;
 private final Counter running;
 private final Meter completed;
 private final Timer idle;
 private final Timer duration;
 /**
 * Wraps an {@link ExecutorService} uses an auto-generated default name.
 *
 * @param delegate {@link ExecutorService} to wrap.
 * @param registry {@link MetricRegistry} that will contain the metrics.
 */
 public InstrumentedExecutorService(ExecutorService delegate, MetricRegistry registry) {
 this(delegate, registry, "instrumented-delegate-" + NAME_COUNTER.incrementAndGet());
 }
 /**
 * Wraps an {@link ExecutorService} with an explicit name.
 *
 * @param delegate {@link ExecutorService} to wrap.
 * @param registry {@link MetricRegistry} that will contain the metrics.
 * @param name name for this executor service.
 */
 public InstrumentedExecutorService(ExecutorService delegate, MetricRegistry registry, String name) {
 this.delegate = delegate;
 this.submitted = registry.meter(MetricRegistry.name(name, "submitted"));
 this.running = registry.counter(MetricRegistry.name(name, "running"));
 this.completed = registry.meter(MetricRegistry.name(name, "completed"));
 this.idle = registry.timer(MetricRegistry.name(name, "idle"));
 this.duration = registry.timer(MetricRegistry.name(name, "duration"));
 }
 @Override
 public void execute(Runnable runnable) {
 submitted.mark();
 delegate.execute(new InstrumentedRunnable(runnable));
 }
 @Override
 public Future<?> submit(Runnable runnable) {
 submitted.mark();
 return delegate.submit(new InstrumentedRunnable(runnable));
 }
 @Override
 public <T> Future<T> submit(Runnable runnable, T result) {
 submitted.mark();
 return delegate.submit(new InstrumentedRunnable(runnable), result);
 }
 @Override
 public <T> Future<T> submit(Callable<T> task) {
 submitted.mark();
 return delegate.submit(new InstrumentedCallable<>(task));
 }
 @Override
 public <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks) throws InterruptedException {
 submitted.mark(tasks.size());
 Collection<? extends Callable<T>> instrumented = instrument(tasks);
 return delegate.invokeAll(instrumented);
 }
 @Override
 public <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks, long timeout, TimeUnit unit) throws InterruptedException {
 submitted.mark(tasks.size());
 Collection<? extends Callable<T>> instrumented = instrument(tasks);
 return delegate.invokeAll(instrumented, timeout, unit);
 }
 @Override
 public <T> T invokeAny(Collection<? extends Callable<T>> tasks) throws ExecutionException, InterruptedException {
 submitted.mark(tasks.size());
 Collection<? extends Callable<T>> instrumented = instrument(tasks);
 return delegate.invokeAny(instrumented);
 }
 @Override
 public <T> T invokeAny(Collection<? extends Callable<T>> tasks, long timeout, TimeUnit unit) throws ExecutionException, InterruptedException, TimeoutException {
 submitted.mark(tasks.size());
 Collection<? extends Callable<T>> instrumented = instrument(tasks);
 return delegate.invokeAny(instrumented, timeout, unit);
 }
 private <T> Collection<? extends Callable<T>> instrument(Collection<? extends Callable<T>> tasks) {
 final List<InstrumentedCallable<T>> instrumented = new ArrayList<>(tasks.size());
 for (Callable<T> task : tasks) {
 instrumented.add(new InstrumentedCallable<>(task));
 }
 return instrumented;
 }
 @Override
 public void shutdown() {
 delegate.shutdown();
 }
 @Override
 public List<Runnable> shutdownNow() {
 return delegate.shutdownNow();
 }
 @Override
 public boolean isShutdown() {
 return delegate.isShutdown();
 }
 @Override
 public boolean isTerminated() {
 return delegate.isTerminated();
 }
 @Override
 public boolean awaitTermination(long l, TimeUnit timeUnit) throws InterruptedException {
 return delegate.awaitTermination(l, timeUnit);
 }
 //......
} 

• InstrumentedExecutorService实现了ExecutorService，对jdk原始的ExecutorService进行了包装，对相应的方法织入指标统计
• 主要统计了已提交的任务submitted(Meter)，运行中的任务running(Counter)，完成的任务completed(Meter)，空闲时长idle(Timer)，运行时长duration(Timer)
• 为了统计后面几个指标，需要对Runnable以及Callable进行织入，因而引入了InstrumentedRunnable、InstrumentedCallable

InstrumentedRunnable

 private class InstrumentedRunnable implements Runnable {
 private final Runnable task;
 private final Timer.Context idleContext;
 InstrumentedRunnable(Runnable task) {
 this.task = task;
 this.idleContext = idle.time();
 }
 @Override
 public void run() {
 idleContext.stop();
 running.inc();
 final Timer.Context durationContext = duration.time();
 try {
 task.run();
 } finally {
 durationContext.stop();
 running.dec();
 completed.mark();
 }
 }
 }

• 织入了对idle、duration、running、completed的统计

InstrumentedCallable

 private class InstrumentedCallable<T> implements Callable<T> {
 private final Callable<T> callable;
 InstrumentedCallable(Callable<T> callable) {
 this.callable = callable;
 }
 @Override
 public T call() throws Exception {
 running.inc();
 final Timer.Context context = duration.time();
 try {
 return callable.call();
 } finally {
 context.stop();
 running.dec();
 completed.mark();
 }
 }
 }

• 织入了对duration、running、completed的统计

ExecutorServiceMetrics

micrometer-core-1.0.3-sources.jar!/io/micrometer/core/instrument/binder/jvm/ExecutorServiceMetrics.java

/**
 * Monitors the status of executor service pools. Does not record timings on operations executed in the {@link ExecutorService},
 * as this requires the instance to be wrapped. Timings are provided separately by wrapping the executor service
 * with {@link TimedExecutorService}.
 *
 * @author Jon Schneider
 * @author Clint Checketts
 */
@NonNullApi
@NonNullFields
public class ExecutorServiceMetrics implements MeterBinder {
 @Nullable
 private final ExecutorService executorService;
 private final Iterable<Tag> tags;
 public ExecutorServiceMetrics(@Nullable ExecutorService executorService, String executorServiceName, Iterable<Tag> tags) {
 this.executorService = executorService;
 this.tags = Tags.concat(tags, "name", executorServiceName);
 }
 //......
 /**
 * Record metrics on the use of an {@link Executor}.
 *
 * @param registry The registry to bind metrics to.
 * @param executor The executor to instrument.
 * @param executorName Will be used to tag metrics with "name".
 * @param tags Tags to apply to all recorded metrics.
 * @return The instrumented executor, proxied.
 */
 public static Executor monitor(MeterRegistry registry, Executor executor, String executorName, Iterable<Tag> tags) {
 if (executor instanceof ExecutorService) {
 return monitor(registry, (ExecutorService) executor, executorName, tags);
 }
 return new TimedExecutor(registry, executor, executorName, tags);
 }
 /**
 * Record metrics on the use of an {@link ExecutorService}.
 *
 * @param registry The registry to bind metrics to.
 * @param executor The executor to instrument.
 * @param executorServiceName Will be used to tag metrics with "name".
 * @param tags Tags to apply to all recorded metrics.
 * @return The instrumented executor, proxied.
 */
 public static ExecutorService monitor(MeterRegistry registry, ExecutorService executor, String executorServiceName, Iterable<Tag> tags) {
 new ExecutorServiceMetrics(executor, executorServiceName, tags).bindTo(registry);
 return new TimedExecutorService(registry, executor, executorServiceName, tags);
 }
 @Override
 public void bindTo(MeterRegistry registry) {
 if (executorService == null) {
 return;
 }
 String className = executorService.getClass().getName();
 if (executorService instanceof ThreadPoolExecutor) {
 monitor(registry, (ThreadPoolExecutor) executorService);
 } else if (className.equals("java.util.concurrent.Executors$DelegatedScheduledExecutorService")) {
 monitor(registry, unwrapThreadPoolExecutor(executorService, executorService.getClass()));
 } else if (className.equals("java.util.concurrent.Executors$FinalizableDelegatedExecutorService")) {
 monitor(registry, unwrapThreadPoolExecutor(executorService, executorService.getClass().getSuperclass()));
 } else if (executorService instanceof ForkJoinPool) {
 monitor(registry, (ForkJoinPool) executorService);
 }
 }
 private void monitor(MeterRegistry registry, @Nullable ThreadPoolExecutor tp) {
 if (tp == null) {
 return;
 }
 FunctionCounter.builder("executor.completed", tp, ThreadPoolExecutor::getCompletedTaskCount)
 .tags(tags)
 .description("The approximate total number of tasks that have completed execution")
 .baseUnit("tasks")
 .register(registry);
 Gauge.builder("executor.active", tp, ThreadPoolExecutor::getActiveCount)
 .tags(tags)
 .description("The approximate number of threads that are actively executing tasks")
 .baseUnit("threads")
 .register(registry);
 Gauge.builder("executor.queued", tp, tpRef -> tpRef.getQueue().size())
 .tags(tags)
 .description("The approximate number of threads that are queued for execution")
 .baseUnit("threads")
 .register(registry);
 Gauge.builder("executor.pool.size", tp, ThreadPoolExecutor::getPoolSize)
 .tags(tags)
 .description("The current number of threads in the pool")
 .baseUnit("threads")
 .register(registry);
 }
 //......
}

• ExecutorServiceMetrics实现了MeterBinder接口，另外提供了静态方法来创建带有监控指标的ExecutorService，该静态方法命名为monitor，非常形象
• monitor方法首先创建ExecutorServiceMetrics，并bindTo了MeterRegistry，然后返回TimedExecutorService
• bindTo方法上报了executor.completed(FunctionCounter)，executor.active(Gauge)，executor.queued(Gauge)，executor.pool.size(Gauge)这几个指标

TimedExecutorService

micrometer-core-1.0.3-sources.jar!/io/micrometer/core/instrument/internal/TimedExecutorService.java

/**
 * An {@link java.util.concurrent.ExecutorService} that is timed
 *
 * @author Jon Schneider
 */
public class TimedExecutorService implements ExecutorService {
 private final ExecutorService delegate;
 private final Timer timer;
 public TimedExecutorService(MeterRegistry registry, ExecutorService delegate, String executorServiceName, Iterable<Tag> tags) {
 this.delegate = delegate;
 this.timer = registry.timer("executor", Tags.concat(tags ,"name", executorServiceName));
 }
 @Override
 public void shutdown() {
 delegate.shutdown();
 }
 @Override
 public List<Runnable> shutdownNow() {
 return delegate.shutdownNow();
 }
 @Override
 public boolean isShutdown() {
 return delegate.isShutdown();
 }
 @Override
 public boolean isTerminated() {
 return delegate.isTerminated();
 }
 @Override
 public boolean awaitTermination(long timeout, TimeUnit unit) throws InterruptedException {
 return delegate.awaitTermination(timeout, unit);
 }
 @Override
 public <T> Future<T> submit(Callable<T> task) {
 return delegate.submit(timer.wrap(task));
 }
 @Override
 public <T> Future<T> submit(Runnable task, T result) {
 return delegate.submit(() -> timer.record(task), result);
 }
 @Override
 public Future<?> submit(Runnable task) {
 return delegate.submit(() -> timer.record(task));
 }
 @Override
 public <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks) throws InterruptedException {
 return delegate.invokeAll(wrapAll(tasks));
 }
 @Override
 public <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks, long timeout, TimeUnit unit) throws InterruptedException {
 return delegate.invokeAll(wrapAll(tasks), timeout, unit);
 }
 @Override
 public <T> T invokeAny(Collection<? extends Callable<T>> tasks) throws InterruptedException, ExecutionException {
 return delegate.invokeAny(wrapAll(tasks));
 }
 @Override
 public <T> T invokeAny(Collection<? extends Callable<T>> tasks, long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException {
 return delegate.invokeAny(wrapAll(tasks), timeout, unit);
 }
 @Override
 public void execute(Runnable command) {
 delegate.execute(timer.wrap(command));
 }
 private <T> Collection<? extends Callable<T>> wrapAll(Collection<? extends Callable<T>> tasks) {
 return tasks.stream().map(timer::wrap).collect(toList());
 }
}

• 对ExecutorService进行包装，增加了

Timer.record

micrometer-core-1.0.3-sources.jar!/io/micrometer/core/instrument/Timer.java

 /**
 * Executes the runnable `f` and records the time taken.
 *
 * @param f Function to execute and measure the execution time.
 */
 void record(Runnable f);
 /**
 * Wrap a {@link Runnable} so that it is timed when invoked.
 *
 * @param f The Runnable to time when it is invoked.
 * @return The wrapped Runnable.
 */
 default Runnable wrap(Runnable f) {
 return () -> record(f);
 }
 /**
 * Wrap a {@link Callable} so that it is timed when invoked.
 *
 * @param f The Callable to time when it is invoked.
 * @param <T> The return type of the callable.
 * @return The wrapped callable.
 */
 default <T> Callable<T> wrap(Callable<T> f) {
 return () -> recordCallable(f);
 }

• warp方法主要是包装调用record方法，而record由实现类去实现

AbstractTimer

micrometer-core-1.0.3-sources.jar!/io/micrometer/core/instrument/AbstractTimer.java

 @Override
 public void record(Runnable f) {
 final long s = clock.monotonicTime();
 try {
 f.run();
 } finally {
 final long e = clock.monotonicTime();
 record(e - s, TimeUnit.NANOSECONDS);
 }
 }
 @Override
 public final void record(long amount, TimeUnit unit) {
 if (amount >= 0) {
 histogram.recordLong(TimeUnit.NANOSECONDS.convert(amount, unit));
 recordNonNegative(amount, unit);
 if (intervalEstimator != null) {
 intervalEstimator.recordInterval(clock.monotonicTime());
 }
 }
 }

• record采用histogram进行统计

小结

dropwizard及micrometer均提供了对ExecutorService的指标统计的包装，micrometer则更近一步提供了静态方法来直接创建，非常方便。