java - ExecutorService - 上下文被销毁时无法从 ServletContextListener 停止线程

Question

我正在开始 Thread来自 ServletContextListener当上下文被初始化并试图在上下文被销毁时停止它。该类是:

public enum BlinkLedTask {

    INSTANCE;

    private Logger logger = RpiLogger.getLogger(getClass());

    private Task task;
    private ExecutorService service;

    private BlinkLedTask() {

    }

    public void run(String[] frequency) {
        stop();

        task = new Task(frequency);
        service = Executors.newSingleThreadExecutor(RpiThreadFactory.INSTANCE);
        service.execute(task);
    }

    public void stop() {
        if(Objects.isNull(task) || Objects.isNull(service)) {
            return;
        }

        try {
            task.terminate();
            service.shutdownNow();
        } catch (Exception cause) {
            logger.error(cause.getMessage(), cause);
        }       
    }

    private static class Task implements Runnable {

        private volatile boolean running = true;
        private String[] frequency;
        private volatile Logger logger = RpiLogger.getLogger(getClass());

        private Task(String[] frequency) {
            this.frequency = frequency;
        }       

        @Override
        public void run() {
            while(running && !Thread.interrupted()) {
                try {
                    resetLed();
                    blinkLed();
                } catch (Throwable cause) {
                    logger.error(cause.getMessage(), cause);
                    running = false;

                    try {
                        resetLed();             
                    } catch (Throwable ignore) {
                    }
                } 
            }
        }

        private void resetLed() throws Exception {
            executeScript(Script.BLINK_LED_RESET);      
        }

        private void blinkLed() throws Exception {
            executeScript(Script.BLINK_LED, new String[]{frequency[0], frequency[1], frequency[2]});        
        }

        private void executeScript(Script script, String... args) {
            ScriptExecutor scriptExecutor = new ScriptExecutor(ScriptExecutor.BASH, script);
            scriptExecutor.execute(true, args);
        }

        private void terminate() {
            logger.info("Stopping - " + Thread.currentThread().getName());
            running = false;
        }
    }
}

这是一个Singleton并且它运行一个 shell 脚本直到它被停止。这个类可以从任何地方调用，所以我需要停止线程，如果有任何当前正在执行 shell 脚本，然后再创建一个新的 Thread。 .

出于测试目的，我执行了 run()初始化上下文并调用 stop() 时此类的方法在毁灭的时候。

删除代码后我重新部署了 war 文件 run() , 我期待 stop()将终止 task ，但它没有。

我还尝试了 run() 的不同实现方式和 stop() :

public void run(String[] frequency) {
    stop();

    task = new Task(frequency);
    Thread thread = RpiThreadFactory.INSTANCE.newThread(task);
    tasks.add(ImmutablePair.of(thread, task));
    thread.start();
}

public void stop() {
    for(ImmutablePair<Thread, Task> pair : tasks) {
        try {
            pair.right.terminate();
            pair.left.join();
        } catch (Exception ex) {

        }           
    }
}

这里是 tasks是private ArrayList<ImmutablePair<Thread, Task>> tasks = new ArrayList<ImmutablePair<Thread,Task>>(); . ImmutablePair属于commons-lang3。但是我收到了java.util.ConcurrentModificationException关于增强的 for 循环的迭代。原因我不知道。

更新

当服务器关闭时 stop()正在按预期工作。我正在使用 Jetty。

更新

RpiThreadFactory :

import java.util.concurrent.ThreadFactory;
import java.util.concurrent.atomic.AtomicInteger;

import org.apache.log4j.Logger;

import com.edfx.rpi.app.utils.logger.RpiLogger;

public enum RpiThreadFactory implements ThreadFactory {
    INSTANCE;

    private final AtomicInteger poolNumber = new AtomicInteger(1);
    private final Logger logger = RpiLogger.getLogger(getClass());
    private final ThreadGroup threadGroup;
    private final AtomicInteger threadNumber = new AtomicInteger(1);
    private final String namePrefix;

    private RpiThreadFactory() {
        SecurityManager securityManager = System.getSecurityManager();
        threadGroup = (securityManager != null) ? securityManager.getThreadGroup() : Thread.currentThread().getThreadGroup();
        namePrefix = "RpiPool-" + poolNumber.getAndIncrement() + "-Thread-";

    }

    public Thread newThread(Runnable runnable) {
        Thread thread = new Thread(threadGroup, runnable, namePrefix + threadNumber.getAndIncrement(), 0);
        thread.setPriority(Thread.NORM_PRIORITY);
        thread.setUncaughtExceptionHandler(new Thread.UncaughtExceptionHandler() {

            public void uncaughtException(Thread thread, Throwable cause) {
                logger.error(cause.getMessage(), cause);
            }
        });

        return thread;
    }
}

ScriptExecutor :

import java.io.BufferedReader;
import java.io.InputStream;
import java.io.InputStreamReader;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.Objects;

import org.apache.commons.lang3.StringUtils;
import org.apache.log4j.Logger;

import com.edfx.rpi.app.utils.logger.RpiLogger;

public class ScriptExecutor {

    private static final Logger LOGGER = RpiLogger.getLogger(ScriptExecutor.class);
    public static final String BASH = "/bin/bash";

    private Script script;
    private Process process;
    private String output;
    private int exitValue;

    public ScriptExecutor(Script script) {
        this.script = script;

    }

    public void execute(boolean destroyProcess, String... args) throws ScriptNotExistException {                
        if(!script.exists()) {
            throw new ScriptNotExistException(script.getScriptName() + " does not exists.");
        }

        try {
            List<String> commands = new ArrayList<>();

            commands.add(BASH);
            commands.add(script.getAbsoultePath());

            if(Objects.nonNull(args)) {
                commands.addAll(Arrays.asList(args));
            }

            StringBuilder builder = new StringBuilder("Executing script: ");
            builder.append(script.getScriptName());

            if(Objects.nonNull(args) && args.length > 0) {
                builder.append(" with parameters: ");
                builder.append(StringUtils.join(args, " "));
            }

            LOGGER.info(builder.toString());

            ProcessBuilder processBuilder = new ProcessBuilder(commands.toArray(new String[commands.size()]));
            process = processBuilder.start();

            StringBuilder outputBuilder = new StringBuilder();
            InputStream inputStream = process.getInputStream();
            InputStreamReader inputStreamReader = new InputStreamReader(inputStream);
            BufferedReader bufferedReader = new BufferedReader(inputStreamReader);

            String line = StringUtils.EMPTY;

            while ((line = bufferedReader.readLine()) != null) {
                outputBuilder.append(line);
                outputBuilder.append("\n");
            }

            process.waitFor();

            exitValue = process.exitValue();
            LOGGER.info("Process for: " + script.getScriptName() + " is executed. Exit value: " + exitValue);

            if(destroyProcess) {
                destroyProcess();
            }

            output = outputBuilder.toString();
        } catch (Exception cause) {
            throw new ScriptExecutionException(cause);
        }       
    }

    public String getOutput() {
        return output;
    }

    public int getExitValue() {
        return exitValue;
    }

    public void destroyProcess() {
        if(Objects.nonNull(process)) {
            LOGGER.info("Process for: " + script.getScriptName() + " is destroyed.");
            process.destroy();
        }
    }
}

目的

这是一个在 Jetty 网络容器中运行的网络应用程序。服务器安装在启用了 java 的嵌入式硬件中。该硬件如何连接 LED。该应用程序接受外部请求，可以是 REST 并启动和停止 LED。因此 LED 可以针对任何请求开始闪烁；但它一次只处理一个请求。

这就是为什么我有 stop如果有的话，它会停止以前运行的进程。 stop适用于正常条件。

但是我看到当 LED 闪烁并且我在没有停止服务器的情况下进行部署时，正在运行的线程并没有停止。如果我停止服务器并进行部署并再次启动，此时正在运行的线程将终止。

while 中的线程循环并执行 Process给本地人。这Process是一次性工作，所以这 Process不会让线程被杀死。

为了重现我所做的问题，我在上下文初始化时创建了线程，并在它被销毁时试图杀死它。现在，如果我在 contextDestroyed 中写点什么我可以看到他们被处决。

我不明白为什么停止服务器不会在我重新部署时杀死线程。

Answer 1

您应该在 processBuilder.start() 返回的 Process 实例上调用 process.destroy()。实际上你在调用 BlinkLedTask.terminate() 时所做的只是设置一些标志。此时您应该调用 process.destroy()。

下面我将展示一个如何重写它的例子。它不涉及你的类 ScriptExecutor(当然你可以将你的逻辑移到那里并在调用 blinkLed() 时将进程实例返回给 BlinkLedTask)。

这里的主要区别在于，我在字段 blinkLedProcess 中保留对 Process 实例的引用，当调用 terminate() 时，我直接调用 process.destroy() 来销毁进程。

您写道“当服务器关闭时，stop() 会按预期工作。我正在使用 Jetty。”确实是的。这是因为通过调用 processBuilder.start(); 您可以创建主 jetty 进程的子进程。当你杀死 jetty 时，它的所有子进程也会被杀死。如果您不终止 jetty，则需要通过调用 destroy() 方法手动终止子进程。

应该是这样的:

public enum BlinkLedTask {
(...)

    private Process resetLedProcess;
    private Process blinkLedProcess;

(...)
   private void blinkLed() throws Exception {
      String[] args = new String[] { frequency[0], frequency[1], frequency[2] };

      List<String> commands = new ArrayList<>();

      //commands.add(BASH);
      commands.add(script.getAbsoultePath());

      if (Objects.nonNull(args)) {
        commands.addAll(Arrays.asList(args));
      }

      StringBuilder builder = new StringBuilder("Executing script: ");
      builder.append(script.getAbsoultePath());

      if (Objects.nonNull(args) && (args.length > 0)) {
        builder.append(" with parameters: ");
        builder.append(StringUtils.join(args, " "));
      }


      ProcessBuilder processBuilder = new ProcessBuilder(commands.toArray(new String[commands.size()]));

      blinkLedProcess = processBuilder.start();

      StringBuilder outputBuilder = new StringBuilder();
      InputStream inputStream = blinkLedProcess.getInputStream();
      InputStreamReader inputStreamReader = new InputStreamReader(inputStream);
      BufferedReader bufferedReader = new BufferedReader(inputStreamReader);

      String line = StringUtils.EMPTY;

      while ((line = bufferedReader.readLine()) != null) {
        outputBuilder.append(line);
        outputBuilder.append("\n");
      }


      blinkLedProcess.waitFor();

      int exitValue = blinkLedProcess.exitValue();
      System.out.println(
        "Process for: " + Script.BLINK_LED.getAbsoultePath() + " is executed. Exit value: " + exitValue);


    }

(...)

   private void terminate() {
      System.out.println("Stopping - " + Thread.currentThread().getName());
      running = false;
      if (resetLedProcess != null) {
        resetLedProcess.destroy();
        System.out.println("Destroyed reset process");
      }
      if (blinkLedProcess != null) {
        blinkLedProcess.destroy();
        System.out.println("Destroyed blink process");
      }
    }
(...)
}