java开发ExecutorService监控实现示例详解

 更新时间:2023年07月10日 09:00:07   作者:codecraft  
这篇文章主要为大家介绍了java开发ExecutorService监控实现示例详解,有需要的朋友可以借鉴参考下,希望能够有所帮助,祝大家多多进步,早日升职加薪

本文主要研究一下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 &gt;= 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

以上就是java开发ExecutorService监控实现示例详解的详细内容,更多关于ExecutorService监控的资料请关注脚本之家其它相关文章!

相关文章

  • Java深入讲解Object类常用方法的使用

    Java深入讲解Object类常用方法的使用

    前面java继承中说到的Object类是java中一个特殊的类,所有的类都是直接或者间接的继承自Object类,即如果某个类没有使用extends关键字则默认是java.lang.Object类的子类,所以所有的类都可以使用Object类中定义的方法,下面介绍Object类的常用方法
    2022-04-04
  • SpringBoot使用validation做参数校验的实现步骤

    SpringBoot使用validation做参数校验的实现步骤

    这篇文章主要介绍了SpringBoot使用validation做参数校验的实现步骤,帮助大家更好的理解和学习使用SpringBoot,感兴趣的朋友可以了解下
    2021-05-05
  • java 异常被catch后 将会继续执行的操作

    java 异常被catch后 将会继续执行的操作

    这篇文章主要介绍了java 异常被catch后 将会继续执行的操作,具有很好的参考价值,希望对大家有所帮助。一起跟随小编过来看看吧
    2021-02-02
  • Spring中使用腾讯云发送短信验证码的实现示例

    Spring中使用腾讯云发送短信验证码的实现示例

    本文主要介绍了Spring 中 使用腾讯云发送短信验证码,文中通过示例代码介绍的非常详细,具有一定的参考价值,感兴趣的小伙伴们可以参考一下
    2022-03-03
  • Spring导入properties配置文件代码示例

    Spring导入properties配置文件代码示例

    这篇文章主要介绍了Spring导入properties配置文件代码示例,文中通过示例代码介绍的非常详细,对大家的学习或者工作具有一定的参考学习价值,需要的朋友可以参考下
    2020-10-10
  • SpringBoot+Querydsl 框架实现复杂查询解析

    SpringBoot+Querydsl 框架实现复杂查询解析

    本篇主要将介绍的是利用spring query dsl框架实现的服务端查询解析和实现介绍,对SpringBoot Querydsl 查询操作感兴趣的朋友一起看看吧
    2022-05-05
  • 将idea工程打包成jar文件的全步骤

    将idea工程打包成jar文件的全步骤

    这篇文章主要给大家介绍了关于将idea工程打包成jar文件的相关资料,文中通过示例代码介绍的非常详细,对大家的学习或者工作具有一定的参考学习价值,需要的朋友们下面来一起学习学习吧
    2019-09-09
  • Java Swing中JDialog实现用户登陆UI示例

    Java Swing中JDialog实现用户登陆UI示例

    这篇文章主要介绍了Java Swing中JDialog实现用户登陆UI功能,结合完整实例形式分析了Swing使用JDialog实现用户登陆UI界面窗口功能的步骤与相关操作技巧,需要的朋友可以参考下
    2017-11-11
  • java 关键字static详细介绍及如何使用

    java 关键字static详细介绍及如何使用

    这篇文章主要介绍了java 关键字static详细介绍及如何使用的相关资料,需要的朋友可以参考下
    2017-03-03
  • Java Excel Poi字体颜色自定义设置代码

    Java Excel Poi字体颜色自定义设置代码

    最近项目使用POI按模板导出Excel,需要设置单元格的字体为红色,下面这篇文章主要给大家介绍了关于Java Excel Poi字体颜色自定义设置的相关资料,需要的朋友可以参考下
    2024-01-01

最新评论