java - 计算工作线程处理消息所花费的时间

Question

检查我的工作线程自拾取消息进行处理以来已经运行了多长时间的最佳方法是什么，然后在超过阈值时间限制时记录错误消息。我认为需要在 WorkerManager 类中进行管理。

1. 我的 WorkerManager 启动工作线程
2. 如果有来自提供者的消息，则工作线程会通过调用服务类来处理这些消息。
3. 如果没有消息，则会短暂 hibernate 。

当我的工作线程处理消息时，如果处理时间超过 5 分钟，那么我想生成一条警告消息，但仍让工作线程继续处理。

问题

我想不断检查我的工作线程处理消息的时间是否超过 5 分钟，如果超过阈值时间，那么我想记录一条错误消息，但仍让工作线程按原样继续。

WorkerManager 类

public class WorkerManager implements Runnable {

    private MyWorker[] workers;
    private int workerCount;
    private boolean stopRequested;

    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();
                }
                managerStarted = true;
            }
        }
    }

    public void stop(){
        stopRequested = true;
    }

    //Calll this
    public void cleanUpOnExit() {
        for(MyWorker w: workers){
            w.setStopRequested();
        }
    }
}

worker 类(Class)

   public class MyWorker implements Runnable {

    private final int WAIT_INTERVAL = 200;
    private MyService myService;
    private MyProvider myProvider;
    private boolean stopRequested = false;

    public MyWorker(MyService myService, MyProvider myProvider){
        this.myService = myService;
        this.myProvider = myProvider;
    }

    public void setStopRequested() {
        stopRequested = true;
    }

    @Override
    public void run() {

        while (!stopRequested) {
            boolean processedMessage = false;
            List<Message> messages = myProvider.getPendingMessages();
            if (messages.size() != 0) {
                AdapterLog.debug("We have " + messages.size() + " messages");
                processedMessage = true;
                for (Message message : messages) {
                    processMessage(messages);
                }
            }

            if (!(processedMessage || 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);
    }
}

Answer 1

您的WorkerManager需要一种方法来确定每个工作人员的最后一条消息何时被处理。因此，工作人员需要跟踪最后处理的消息的时间戳。

然后，您的 WorkerManager 可以检查每个工作人员的时间戳并在需要时生成警告。为了检查给定时间段内的工作人员，您可以使用执行器:

    ScheduledExecutorService scheduledExecutorService = Executors.newSingleThreadScheduledExecutor();
    scheduledExecutorService.scheduleAtFixedRate(this::checkTimeoutProcessingMessages, 5l, 5l, TimeUnit.SECONDS);

您可以检查从每个工作人员获取时间戳的时间:

    public void checkTimeoutProcessingMessages() {
        for (MyWorker worker : workers) {
            long lastProcessed = worker.getLastProcessedMessageTimestamp();
            long currentTimestamp = System.currentTimeMillis();
            if (lastProcessed + 5000 > currentTimestamp) {
                //warn message
            }
        }
    }