聊聊ExecutorService的监控

青旅半醒 2022-03-16 12:58 193阅读 0赞

## 序 ##

本文主要研究一下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则更近一步提供了静态方法来直接创建，非常方便。

## doc ##

*  [InstrumentedExecutorService][]
 *  [ExecutorServiceMetrics][]

[InstrumentedExecutorService]: https://link.juejin.im?target=https%3A%2F%2Fgithub.com%2Fdropwizard%2Fmetrics%2Fblob%2F4.1-development%2Fmetrics-core%2Fsrc%2Fmain%2Fjava%2Fcom%2Fcodahale%2Fmetrics%2FInstrumentedExecutorService.java
[ExecutorServiceMetrics]: https://link.juejin.im?target=https%3A%2F%2Fgithub.com%2Fmicrometer-metrics%2Fmicrometer%2Fblob%2Fmaster%2Fmicrometer-core%2Fsrc%2Fmain%2Fjava%2Fio%2Fmicrometer%2Fcore%2Finstrument%2Fbinder%2Fjvm%2FExecutorServiceMetrics.java