具有持久资源的Java线程

Question

我有一个工作线程池(一个ExecutorService)。

此池用于运行shell命令。

我使用shell (/bin/sh)，而不是直接为可执行文件创建进程，因为我使用shell重定向(>)将输出直接写入磁盘，而不必通过JVM，以及其他一些细节。

生成shell进程需要2-3毫秒。

我希望每个线程保持一个shell进程，以避免启动它的开销。

如何允许每个线程拥有一个进程？

我正在考虑在线程局部变量中使用ThreadFactory。

class ThreadFactory {
    Thread newThread(Runnable r) {
        return new Thread(new Runnable() {
            Process process = Runtime.getRuntime().exec("/bin/sh")
            try {
                // store process as thread local here
                r.run(); // then r can access thread local
            } catch(Exception e) {
                try {
                    process.close();
                } catch(Exception e) {
                }
                throw e;
            }
        });
    }
}

(或者，我可以在我的Thread中将Thread.currentThread()子类转换为该类。)

这是解决这个问题的好办法吗？

Answer 1

我会将Process引用保存在一个连续执行命令的ProcessRunnable中。我认为比使用ThreadLocal和ThreadFactory更清晰。就像这样：

public class ShellCommandExecutor {

private int concurrency = 10;
private int capacity = 100;

private ExecutorService service = Executors.newFixedThreadPool(concurrency);
private BlockingQueue<String> commandsQueue = new LinkedBlockingQueue<>(capacity);

public void start() {
    for (int i = 0; i < concurrency; i++)
        service.submit(new Runnable() {
            @Override
            public void run() {
                //todo deal with ioexception
                Process process = Runtime.getRuntime().exec("/bin/sh");
                while (!Thread.currentThread().isInterrupted()) {
                    try {
                        String command = commandsQueue.take();
                        //todo execute commands using the same process per thread
                    }
                    catch (InterruptedException e) {
                        Thread.currentThread().interrupt();
                    }
                }
            }
        });
}

public void executeCommand(String command) throws InterruptedException {
    commandsQueue.put(command);
}

public void shutdown() {
    service.shutdownNow();
}
}

编辑：一种具有线程本地的解决方案，可以轻松地处理缓存的线程池：

public class ShellCommandExecutor2 {

    //todo limit queue
    private ExecutorService service = Executors.newCachedThreadPool();

    public void executeCommand(final String command) throws InterruptedException {

        service.submit(new Runnable() {
            @Override
            public void run() {
                Process process = ThreadLocalProcessFactory.get();
                //todo execute command
            }
        });
    }

    public void shutdown() {
        service.shutdownNow();
    }

    private static class ThreadLocalProcessFactory {

        private static final ThreadLocal<Process> processThreadLocal =
                new ThreadLocal<Process>() {
                    @Override protected Process initialValue() {
                        try {
                            return Runtime.getRuntime().exec("/bin/sh");
                        }
                        catch (IOException e) {
                            e.printStackTrace();
                            return null;
                        }
                    }
                };

        static Process get() {
            return processThreadLocal.get();
        }
    }
}