java - 测量消费者/生产者工作的时间

Question

生产者和消费者的当前配置:

ExecutorService consumerExecutor = createExecutor(...);
ExecutorService producerExecutor = createExecutor(...);

try {
    List<Callable<Integer>> callablesForConsumer = createListOfCallablesForConsumer(...);
    List<Callable<Integer>> callablesForProducer = createListOfCallablesForProducer(...);
    ....
    ....
    // submitting tasks to executors and combine them into one list of futures
    ....
    ....
    for (Future<Integer> future : futures) {
        intCount += future.get();
    }
    ....
    ....
    // some business logic
    ....
    ....
} finally {
    consumerExecutor.shutdown();
    producerExecutor.shutdown();
}

如何测量和记录消费者和生产者分别花费的总时间？

因此我需要得到这样的东西:producerTotalTime=... ms, consumerTotalTime=... ms

我是否应该为它覆盖一些执行程序服务方法，对此有什么想法吗？

Answer 1

您可以覆盖 ThreadPoolExecutor 的两个方法:

beforeExecute(Thread t, Runnable r)和 afterExecute(Runnable r, Throwable t) , 都采用 runnable，因此使用该 runnable 来计算其执行时间(将 runnable 映射到例如 ConcurrentHashMap<Runable, Long> 中)

然后在beforeExecute , 把 Runnable和 System.currentTimeMillis()到 map 。

在afterExecute , 取 startTime从 map Runnable , 并删除此 Runnable从 map 上。然后使用 startTime计算long taskExecutionTime = System.currentTimeMillis() - startTime然后使用 AtomicLong总结所有taskExecutionTime来自每个可运行的。

使用 ConcurrentHashMap您只需要添加新的 Runnable，因为添加相同的 Runnable 两次或更多次将覆盖前一个。如果您想多次添加相同的可运行对象，请使用一些 MultiMap实现(例如来自 commons-collection 或 guava)。这个MultiMap需要并发实现(线程安全)。

另一种选择是使用一些 ThreadLocal 变量而不是 ConcurrentHashMap，因为每个 Runnable 在自己的(执行时刻)线程中执行。在执行前将startTime添加到threadLocal，在afterExecute中获取并计算taskExecutionTime。

第二种情况下的实现可能如下所示:

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;

public class TimedExecutor extends ThreadPoolExecutor {
    private ThreadLocal<Long> startTime = new ThreadLocal<>();
    private AtomicLong totalExecutionTime = new AtomicLong(0);

    public TimedExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit,
            BlockingQueue<Runnable> workQueue) {
        super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue);
    }

    @Override
    protected void beforeExecute(Thread t, Runnable r) {
        super.beforeExecute(t, r);
        startTime.set(System.currentTimeMillis());
    }

    @Override
    protected void afterExecute(Runnable r, Throwable t) {
        super.afterExecute(r, t);
        long taskExecutionTime = System.currentTimeMillis() - startTime.get();
        totalExecutionTime.addAndGet(taskExecutionTime);
    }

    public long totalExecutionTime() {
        return totalExecutionTime.get();
    }

    public static TimedExecutor newFixedThreadPool(int noOfThreads) {
        int corePoolSize = noOfThreads;
        int maximumPoolSize = noOfThreads;
        return new TimedExecutor(corePoolSize, maximumPoolSize, 0L, TimeUnit.MILLISECONDS,
                new LinkedBlockingQueue<Runnable>());
    }
}

如果您想测量特定 ExecutorService 运行和完成某些特定任务集所需的时间，那么这可以给您一个提示:

import java.util.LinkedList;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

public class MeasureExecutorTime {

    public static void main(String[] args) throws InterruptedException {
        ExecutorService executorService = Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors()); // any executor - your choice
        LinkedList<Runnable> runnables = new LinkedList<>(); // place your Runnables here, to minimize influence of initialization.

        long startTime = System.currentTimeMillis();
        for (Runnable command : runnables) {
            executorService.execute(command);
        }
        executorService.shutdownNow();
        // next line will block till all tasks finishes
        executorService.awaitTermination(Long.MAX_VALUE, TimeUnit.DAYS);

        long totalExecutionTime = System.currentTimeMillis() - startTime;
    }
}