java - 从 Callable 启动和停止 Process Thread

Question

我有一个可启动线程的可调用对象(该线程运行一个 ping 进程)，我想允许用户取消任务:

public class PingCallable implements Callable<PingResult> {

private ProcThread processThread;

public PingCallable(String ip) {
    this.processThread = new ProcThread(ip);
}

@Override
public PingResult call() throws Exception {
    log.trace("Checking if the ip " + ip + " is alive");
    try {
        processThread.start();
        try {
            processThread.join();
        } catch (InterruptedException e) {
            log.error("The callable thread was interrupted for " + processThread.getName());
            processThread.interrupt();
            // Good practice to reset the interrupt flag.
            Thread.currentThread().interrupt();
        }
    } catch (Throwable e) {
        System.out.println("Throwable ");
    }
    return new PingResult(ip, processThread.isPingAlive());
  }
}

ProcThread，看起来像:

@Override
public void run() {
    try {
        process = Runtime.getRuntime().exec("the long ping", null, workDirFile);
        /* Get process input and error stream, not here to keep it short*/ 

        // waitFor is InterruptedException sensitive
        exitVal = process.waitFor();
    } catch (InterruptedException ex) {
        log.error("interrupted " + getName(), ex);
        process.destroy();
        /* Stop the intput and error stream handlers, not here */ 
        // Reset the status, good practice
        Thread.currentThread().interrupt();
    } catch (IOException ex) {
        log.error("Exception while execution", ex);
    }
}

测试:

    @Test
    public void test() throws ExecutionException, InterruptedException {
        ExecutorService executorService = Executors.newFixedThreadPool(15);
        List<Future<PingResult>> futures = new ArrayList<>();

        for (int i= 0; i < 100; i++) {
            PingCallable pingTask = new PingCallable("10.1.1.142");
            futures.add(executorService.submit(pingTask));
        }

        Thread.sleep(10000);
        executorService.shutdownNow();
//        for (Future<PingResult> future : futures) {
//            future.cancel(true);
//        }
    }

我使用ProcessExplorer监控ping进程，我看到15，然后执行shutdownNow，或者future.cancel(true)，只有4-5个最多8个进程被中断，其余的都活着，我几乎从来没有看到15消息显示“可调用线程被中断..”，并且测试直到进程结束才完成。 这是为什么？

Answer 1

我可能没有完整的答案，但有两件事需要注意:

• shutdownNow发出关闭信号，要查看线程是否实际停止，请使用 awaitTermination
• process.destroy()执行也需要时间，因此可调用程序应该在中断进程线程后等待其完成。

我稍微修改了一下代码，发现future.cancel(true)实际上会阻止 catch InterruptedException 中任何内容的执行-ProcThread block ，除非您使用 executor.shutdown()而不是executor.shutdownNow() 。当打印“Executor终止:true”时，单元测试完成(使用junit 4.11)。看起来像使用 future.cancel(true)和executor.shutdownNow()将双重中断线程，这可能导致中断 block 被跳过。

下面是我用于测试的代码。取消注释 for (Future<PingResult> f : futures) f.cancel(true);与 shutdown(Now) 一起查看输出的差异。

public class TestRunInterrupt {


static long sleepTime = 1000L;
static long killTime = 2000L;

@Test
public void testInterrupts() throws Exception {

    ExecutorService executorService = Executors.newFixedThreadPool(3);
    List<Future<PingResult>> futures = new ArrayList<Future<PingResult>>();
    for (int i= 0; i < 100; i++) {
        PingCallable pingTask = new PingCallable("10.1.1.142");
        futures.add(executorService.submit(pingTask));
    }
    Thread.sleep(sleepTime + sleepTime / 2);
    // for (Future<PingResult> f : futures) f.cancel(true);
    // executorService.shutdown();
    executorService.shutdownNow();
    int i = 0;
    while (!executorService.isTerminated()) {
        System.out.println("Awaiting executor termination " + i);
        executorService.awaitTermination(1000L, TimeUnit.MILLISECONDS);
        i++;
        if (i > 5) {
            break;
        }
    }
    System.out.println("Executor terminated: " + executorService.isTerminated());
}

static class ProcThread extends Thread {

    static AtomicInteger tcount = new AtomicInteger();

    int id;
    volatile boolean slept;

    public ProcThread() {
        super();
        id = tcount.incrementAndGet();
    }

    @Override
    public void run() {

        try {
            Thread.sleep(sleepTime);
            slept = true;
        } catch (InterruptedException ie) {
            // Catching an interrupted-exception clears the interrupted flag.
            System.out.println(id + " procThread interrupted");
            try {
                Thread.sleep(killTime);
                System.out.println(id + " procThread kill time finished");
            } catch (InterruptedException ie2) {
                System.out.println(id + "procThread killing interrupted"); 
            }
            Thread.currentThread().interrupt();
        } catch (Throwable t) {
            System.out.println(id + " procThread stopped: " + t);
        }
    }
}

static class PingCallable implements Callable<PingResult> {

    ProcThread pthread;

    public PingCallable(String s) {
        pthread = new ProcThread();
    }

    @Override
    public PingResult call() throws Exception {

        System.out.println(pthread.id + " starting sleep");
        pthread.start();
        try {
            System.out.println(pthread.id + " awaiting sleep");
            pthread.join();
        } catch (InterruptedException ie) {
            System.out.println(pthread.id + " callable interrupted");
            pthread.interrupt();
            // wait for kill process to finish
            pthread.join();
            System.out.println(pthread.id + " callable interrupt done");
            Thread.currentThread().interrupt();
        } catch (Throwable t) {
            System.out.println(pthread.id + " callable stopped: " + t);
        }
        return new PingResult(pthread.id, pthread.slept);
    }
}

static class PingResult {

    int id;
    boolean slept;

    public PingResult(int id, boolean slept) {
        this.id = id;
        this.slept = slept;
        System.out.println(id + " slept " + slept);
    }
}

}

没有 future.cancel(true) 的输出或与 future.cancel(true)和正常shutdown() :
1 starting sleep
1 awaiting sleep
2 starting sleep
3 starting sleep
2 awaiting sleep
3 awaiting sleep
1 slept true
3 slept true
2 slept true
5 starting sleep
4 starting sleep
6 starting sleep
5 awaiting sleep
6 awaiting sleep
4 awaiting sleep
4 callable interrupted
Awaiting executor termination 0
6 callable interrupted
4 procThread interrupted
5 callable interrupted
6 procThread interrupted
5 procThread interrupted
Awaiting executor termination 1
6 procThread kill time finished
5 procThread kill time finished
4 procThread kill time finished
5 callable interrupt done
5 slept false
6 callable interrupt done
4 callable interrupt done
6 slept false
4 slept false
Executor terminated: true

输出为 future.cancel(true)和shutdownNow() :
1 starting sleep
2 starting sleep
1 awaiting sleep
2 awaiting sleep
3 starting sleep
3 awaiting sleep
3 slept true
2 slept true
1 slept true
4 starting sleep
6 starting sleep
5 starting sleep
4 awaiting sleep
5 awaiting sleep
6 awaiting sleep
5 callable interrupted
6 callable interrupted
4 callable interrupted
5 procThread interrupted
6 procThread interrupted
4 procThread interrupted
Executor terminated: true