java - 计算我的工作线程仍在处理时所花费的时间

Question

我无法获取工作线程的处理时间。我有一个线程监视工作线程的处理时间，并报告所花费的时间以及有多少工作线程处于处理状态。

问题

1. 我的监控线程，总是报告没有处于处理状态的worker？我怀疑我的监控线程无法正确评估工​​作线程。

2. 我已经同步了一些获取和记录处理时间和处理状态的方法。这足以让我的监视器线程在不覆盖线程的情况下跟踪处理时间吗？

3. isProcessing boolean 变量是否应该是 volatile 变量？

MyWorker 类

public class MyWorker implements Runnable {

    private final int WAIT_INTERVAL = 200;
    private MyService myService;
    private MyProvider myProvider;
    private boolean stopRequested = false;
    private boolean isProcessing; // Flag to indicate the worker is currently processing a message
    private long processingStartTime; //captures when the worker started processing a message
    private Logger logger = new Logger();

public MyWorker(MyService myService, MyProvider myProvider){
        this.myService = myService;
        this.myProvider = myProvider;
        this.isProcessing = false;
        this.processingStartTime = -1;
    }

    public void setStopRequested() {
        stopRequested = true;
    }


    private synchronized void recordProcessingStart(long start){
        isProcessing = true;
        processingStartTime = start;
    }

    private synchronized void recordProcessingStop(){
        isProcessing = false;
        processingStartTime = -1;
    }

    public synchronized boolean isProcessing(){
        return isProcessing;
    }

    public synchronized long getProcessingStartTime(){
        return processingStartTime;
    }

    @Override
    public void run() {

        while (!stopRequested) {
            boolean processingMessage = false;
            List<Message> messages = myProvider.getPendingMessages();
            if (messages.size() != 0) {
                logger.log("We have " + messages.size() + " messages");
                recordProcessingStart(System.currentTimeMillis());
                for (Message message : messages) {
                    processMessage(messages);
                }
                recordProcessingStop();
            }

            if (!(processingMessage || stopRequested)) {
                // this is to stop the thread from spinning when there are no messages
                try {
                    Thread.sleep(WAIT_INTERVAL);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }
    }

    private void processMessage(Message messages){
        myService.process(messages);
    }
}

WorkerManager类

public class WorkerManager implements Runnable {

    private MyWorker[] workers;
    private int workerCount;
    private boolean stopRequested;
    private ScheduledExecutorService scheduledExecutorService = Executors.newSingleThreadScheduledExecutor();
    private Logger logger = new Logger();

    public WorkerManager(int count) {
        this.workerCount = count;
    }

    @Override
    public void run() {
        stopRequested = false;
        boolean managerStarted = false;

        while (!stopRequested) {
            if (!managerStarted) {
                workers = new MyWorker[workerCount];
                for (int i = 0; i < workerCount; i++) {
                    final Thread workerThread = new Thread(workers[i], "Worker-" + (i + 1));
                    workerThread.start();
                }
                startProcessMonitoringThread();
                managerStarted = true;
            }
        }
    }

    public void stop() {
        stopRequested = true;
    }

    public void cleanUpOnExit() {
        for (MyWorker w : workers) {
            w.setStopRequested();
        }
        stopProcessMonitoringThread();
    }

    private void startProcessMonitoringThread(){

        scheduledExecutorService = Executors.newScheduledThreadPool(1);
        scheduledExecutorService.scheduleWithFixedDelay( new WorkerMonitorThread(workers), 1, 1, TimeUnit.MINUTES);
        logger.info("Monitoring of worker threads initialized ...");
    }

    private void stopProcessMonitoringThread(){
        scheduledExecutorService.shutdown();
        logger.info("Successfully shutdown worker monitoring thread ...");
    }


    private class WorkerMonitorThread implements Runnable{

        private MyWorker[] workers;
        WorkerMonitorThread(MyWorker workers[]){
            this.workers = workers;

        }

        @Override
        public void run() {
            String trackingId = "["+ UUID.randomUUID().toString() + "] - ";
            logger.info(trackingId + "Calculating processing times of worker threads ...");

            StringBuilder sb = new StringBuilder();
            int totalWorkersInProcess = 0;
            for (int i = 0; i < workerCount; i++) {
                MyWorker worker = workers[i];
                if(worker.isProcessing()){
                    totalWorkersInProcess++;
                    long startTime = worker.getProcessingStartTime();
                    long currentTime = System.currentTimeMillis();
                    long duration = currentTime - startTime;
                    sb.append(trackingId + "Worker-" + (i + 1) + " has been processing for the last : " + duration + " ms");
                }
            }
            sb.append(trackingId + "Total worker(s) in progress : " + totalWorkersInProcess);
            logger.info(sb.toString());
        }
    }
}

Answer 1

您还没有创建您的工作人员。

查看 WorkerManager run 方法的代码:

            workers = new MyWorker[workerCount]; // Create an array of MyWorker, but each element of the array is null.
            for (int i = 0; i < workerCount; i++) {
                final Thread workerThread = new Thread(workers[i], "Worker-" + (i + 1)); // <-- Creates a thread and pass a null MyWorker.
                workerThread.start(); // NullPointers are coming on the other thread.
            }

要修复该错误，您的 WorkerManager 应该创建工作人员或接收要使用的工作人员。您可以向构造函数添加依赖项来创建工作人员:

    public WorkerManager(int count, MyService myService, MyProvider myProvider) {
        this.workerCount = count;
        this.myService = myService;
        this.myProvider = myProvider;
    }

然后，正确创建 worker :

            workers = new MyWorker[workerCount];
            for (int i = 0; i < workerCount; i++) {
                workers[i] = new MyWorker(myService, myProvider);
            }