【问题标题】:ThreadPool of CLI ProcessesCLI 进程的线程池
【发布时间】:2012-04-05 23:21:00
【问题描述】:

我需要通过 Java 中的标准输入将消息传递给 CLI PHP 进程。我想在一个池中保持大约 20 个 PHP 进程运行,这样当我将消息传递到池时,它会将每条消息发送到一个单独的线程,保留一个要传递的消息队列。我希望这些 PHP 进程尽可能长地保持活力,如果一个进程死了,就会产生一个新进程。我研究过使用静态线程池来执行此操作,但它似乎更适合执行并简单地死掉的任务。我怎么能做到这一点,通过一个简单的界面将消息传递到池?我是否必须实现自己的自定义“线程池”?

【问题讨论】:

  • 非常类似于这个问题:*.com/questions/2592093/php-thread-pool
  • 我有任何来自 PHP 的输出,以便您知道它何时完成处理?
  • 它永远不会被完成处理。如果一个人死了,我需要产生一个新的来替换它。我将通过标准输入以循环方式将数据传递给他们。

标签: java php command-line threadpool stdin


【解决方案1】:

我为此提供了一些代码,因为我认为它会使事情变得更清晰。基本上,您需要保留一个进程对象池。请注意,这些流程中的每一个都有您需要以某种方式管理的输入、输出和错误流。在我的示例中,我只是将错误和输出重定向到主进程控制台。如果需要,您可以设置回调和处理程序以获取 PHP 程序的输出。如果您只是在处理任务并且不在乎 PHP 说什么,请保持原样或重定向到文件。

我正在为 ObjectPool 使用 Apache Commons Pool 库。无需重新发明一个。

您将拥有一个包含 20 个进程的池来运行您的 PHP 程序。仅此一项并不能满足您的需要。您可能希望“同时”针对所有 20 个进程处理任务。所以你还需要一个 ThreadPool 来从你的 ObjectPool 中拉取一个进程。

您还需要了解,如果您杀死或按 CTRL-C 您的 Java 进程,init 进程将接管您的 php 进程,它们将坐在那里。您可能希望记录您生成的 PHP 进程的所有 pid,然后在重新运行 Java 程序时清理它们。

public class *_10037379 {

    private static Logger sLogger = Logger.getLogger(*_10037379.class.getName());

    public static class CLIPoolableObjectFactory extends BasePoolableObjectFactory<Process> {

        private String mProcessToRun;

        public CLIPoolableObjectFactory(String processToRun) {
            mProcessToRun = processToRun;
        }

        @Override
        public Process makeObject() throws Exception {
            ProcessBuilder builder = new ProcessBuilder();
            builder.redirectError(Redirect.INHERIT);
            // I am being lazy, but really the InputStream is where
            // you can get any output of the PHP Process. This setting
            // will make it output to the current processes console.
            builder.redirectOutput(Redirect.INHERIT);
            builder.redirectInput(Redirect.PIPE);
            builder.command(mProcessToRun);
            return builder.start();
        }

        @Override
        public boolean validateObject(Process process) {
            try {
                process.exitValue();
                return false;
            } catch (IllegalThreadStateException ex) {
                return true;
            }
        }

        @Override
        public void destroyObject(Process process) throws Exception {
            // If PHP has a way to stop it, do that instead of destroy
            process.destroy();
        }

        @Override
        public void passivateObject(Process process) throws Exception {
            // Should really try to read from the InputStream of the Process
            // to prevent lock-ups if Rediret.INHERIT is not used.
        }
    }

    public static class CLIWorkItem implements Runnable {

        private ObjectPool<Process> mPool;
        private String mWork;

        public CLIWorkItem(ObjectPool<Process> pool, String work) {
            mPool = pool;
            mWork = work;
        }

        @Override
        public void run() {
            Process workProcess = null;
            try {
                workProcess = mPool.borrowObject();
                OutputStream os = workProcess.getOutputStream();
                os.write(mWork.getBytes(Charset.forName("UTF-8")));
                os.flush();
                // Because of the INHERIT rule with the output stream
                // the console stream overwrites itself. REMOVE THIS in production.
                Thread.sleep(100);
            } catch (Exception ex) {
                sLogger.log(Level.SEVERE, null, ex);
            } finally {
                if (workProcess != null) {
                    try {
                        // Seriously.. so many exceptions.
                        mPool.returnObject(workProcess);
                    } catch (Exception ex) {
                        sLogger.log(Level.SEVERE, null, ex);
                    }
                }
            }
        }
    }

    public static void main(String[] args) throws Exception {

        // Change the 5 to 20 in your case. 
        // Also change mock_php.exe to /usr/bin/php or wherever.
        ObjectPool<Process> pool =
                new GenericObjectPool<>(
                new CLIPoolableObjectFactory("mock_php.exe"), 5);         

        // This will only allow you to queue 100 work items at a time. I would suspect
        // that if you only want 20 PHP processes running at a time and this queue
        // filled up you'll need to implement some other strategy as you are doing
        // more work than PHP can keep up with. You'll need to block at some point
        // or throw work away.
        BlockingQueue<Runnable> queue = 
            new ArrayBlockingQueue<>(100, true);

        ThreadPoolExecutor executor = 
            new ThreadPoolExecutor(20, 20, 1, TimeUnit.HOURS, queue);

        // print some stuff out.
        executor.execute(new CLIWorkItem(pool, "Message 1\r\n"));
        executor.execute(new CLIWorkItem(pool, "Message 2\r\n"));
        executor.execute(new CLIWorkItem(pool, "Message 3\r\n"));
        executor.execute(new CLIWorkItem(pool, "Message 4\r\n"));
        executor.execute(new CLIWorkItem(pool, "Message 5\r\n"));
        executor.execute(new CLIWorkItem(pool, "Message 6\r\n"));
        executor.execute(new CLIWorkItem(pool, "Message 7\r\n"));
        executor.execute(new CLIWorkItem(pool, "Message 8\r\n"));
        executor.execute(new CLIWorkItem(pool, "Message 9\r\n"));
        executor.execute(new CLIWorkItem(pool, "Message 10\r\n"));
        executor.execute(new CLIWorkItem(pool, "Message 11\r\n"));

        executor.shutdown();
        executor.awaitTermination(4000, TimeUnit.HOURS);

        pool.close();        
    }
}

程序运行的输出:

12172 - Message 2
10568 - Message 1
4804 - Message 3
11916 - Message 4
11116 - Message 5
12172 - Message 6
4804 - Message 7
10568 - Message 8
11916 - Message 9
11116 - Message 10
12172 - Message 11

仅输出输入内容的 C++ 程序代码:

#include <windows.h>
#include <iostream>
#include <string>

int main(int argc, char* argv[])
{
    DWORD pid = GetCurrentProcessId();
    std::string line;
    while (true) {      
        std::getline (std::cin, line);
        std::cout << pid << " - " << line << std::endl;
    }

    return 0;
}

更新

抱歉耽搁了。这是任何感兴趣的人的 JDK 6 版本。您必须运行一个单独的线程来从进程的 InputStream 读取所有输入。我已将此代码设置为在每个新进程旁边生成一个新线程。只要该线程还活着,它就总是从进程中读取。我没有直接输出到文件,而是将其设置为使用 Logging 框架。这样,您可以设置日志配置以转到文件、翻转、转到控制台等,而无需硬编码以转到文件。

你会注意到我只为每个进程启动一个 Gobbler,即使一个进程有标准输出和标准错误。我将 stderr 重定向到 stdout 只是为了让事情变得更容易。显然 jdk6 只支持这种类型的重定向。

public class *_10037379_jdk6 {

    private static Logger sLogger = Logger.getLogger(*_10037379_jdk6.class.getName());

    // Shamelessy taken from Google and modified. 
    // I don't know who the original Author is.
    public static class StreamGobbler extends Thread {

        InputStream is;
        Logger logger;
        Level level;

        StreamGobbler(String logName, Level level, InputStream is) {
            this.is = is;
            this.logger = Logger.getLogger(logName);
            this.level = level;
        }

        public void run() {
            try {
                InputStreamReader isr = new InputStreamReader(is);
                BufferedReader br = new BufferedReader(isr);
                String line = null;
                while ((line = br.readLine()) != null) {
                    logger.log(level, line);
                }
            } catch (IOException ex) {
                logger.log(Level.SEVERE, "Failed to read from Process.", ex);
            }
            logger.log(
                    Level.INFO, 
                    String.format("Exiting Gobbler for %s.", logger.getName()));
        }
    }

    public static class CLIPoolableObjectFactory extends BasePoolableObjectFactory<Process> {

        private String mProcessToRun;

        public CLIPoolableObjectFactory(String processToRun) {
            mProcessToRun = processToRun;
        }

        @Override
        public Process makeObject() throws Exception {
            ProcessBuilder builder = new ProcessBuilder();
            builder.redirectErrorStream(true);
            builder.command(mProcessToRun);
            Process process = builder.start();
            StreamGobbler loggingGobbler =
                    new StreamGobbler(
                    String.format("process.%s", process.hashCode()),
                    Level.INFO,
                    process.getInputStream());
            loggingGobbler.start();
            return process;
        }

        @Override
        public boolean validateObject(Process process) {
            try {
                process.exitValue();
                return false;
            } catch (IllegalThreadStateException ex) {
                return true;
            }
        }

        @Override
        public void destroyObject(Process process) throws Exception {
            // If PHP has a way to stop it, do that instead of destroy
            process.destroy();
        }

        @Override
        public void passivateObject(Process process) throws Exception {
            // Should really try to read from the InputStream of the Process
            // to prevent lock-ups if Rediret.INHERIT is not used.
        }
    }

    public static class CLIWorkItem implements Runnable {

        private ObjectPool<Process> mPool;
        private String mWork;

        public CLIWorkItem(ObjectPool<Process> pool, String work) {
            mPool = pool;
            mWork = work;
        }

        @Override
        public void run() {
            Process workProcess = null;
            try {
                workProcess = mPool.borrowObject();
                OutputStream os = workProcess.getOutputStream();
                os.write(mWork.getBytes(Charset.forName("UTF-8")));
                os.flush();
                // Because of the INHERIT rule with the output stream
                // the console stream overwrites itself. REMOVE THIS in production.
                Thread.sleep(100);
            } catch (Exception ex) {
                sLogger.log(Level.SEVERE, null, ex);
            } finally {
                if (workProcess != null) {
                    try {
                        // Seriously.. so many exceptions.
                        mPool.returnObject(workProcess);
                    } catch (Exception ex) {
                        sLogger.log(Level.SEVERE, null, ex);
                    }
                }
            }
        }
    }

    public static void main(String[] args) throws Exception {

        // Change the 5 to 20 in your case. 
        ObjectPool<Process> pool =
                new GenericObjectPool<Process>(
                new CLIPoolableObjectFactory("mock_php.exe"), 5);

        BlockingQueue<Runnable> queue = new ArrayBlockingQueue<Runnable>(100, true);

        ThreadPoolExecutor executor = new ThreadPoolExecutor(20, 20, 1, TimeUnit.HOURS, queue);

        // print some stuff out.
        executor.execute(new CLIWorkItem(pool, "Message 1\r\n"));
        executor.execute(new CLIWorkItem(pool, "Message 2\r\n"));
        executor.execute(new CLIWorkItem(pool, "Message 3\r\n"));
        executor.execute(new CLIWorkItem(pool, "Message 4\r\n"));
        executor.execute(new CLIWorkItem(pool, "Message 5\r\n"));
        executor.execute(new CLIWorkItem(pool, "Message 6\r\n"));
        executor.execute(new CLIWorkItem(pool, "Message 7\r\n"));
        executor.execute(new CLIWorkItem(pool, "Message 8\r\n"));
        executor.execute(new CLIWorkItem(pool, "Message 9\r\n"));
        executor.execute(new CLIWorkItem(pool, "Message 10\r\n"));
        executor.execute(new CLIWorkItem(pool, "Message 11\r\n"));

        executor.shutdown();
        executor.awaitTermination(4000, TimeUnit.HOURS);

        pool.close();
    }
}

输出

Apr 11, 2012 8:41:02 PM *.*_10037379_jdk6$StreamGobbler run
INFO: 9440 - Message 3
Apr 11, 2012 8:41:02 PM *.*_10037379_jdk6$StreamGobbler run
INFO: 8776 - Message 2
Apr 11, 2012 8:41:02 PM *.*_10037379_jdk6$StreamGobbler run
INFO: 6100 - Message 1
Apr 11, 2012 8:41:02 PM *.*_10037379_jdk6$StreamGobbler run
INFO: 10096 - Message 4
Apr 11, 2012 8:41:02 PM *.*_10037379_jdk6$StreamGobbler run
INFO: 8868 - Message 5
Apr 11, 2012 8:41:02 PM *.*_10037379_jdk6$StreamGobbler run
INFO: 8868 - Message 8
Apr 11, 2012 8:41:02 PM *.*_10037379_jdk6$StreamGobbler run
INFO: 6100 - Message 10
Apr 11, 2012 8:41:02 PM *.*_10037379_jdk6$StreamGobbler run
INFO: 8776 - Message 9
Apr 11, 2012 8:41:02 PM *.*_10037379_jdk6$StreamGobbler run
INFO: 10096 - Message 6
Apr 11, 2012 8:41:02 PM *.*_10037379_jdk6$StreamGobbler run
INFO: 9440 - Message 7
Apr 11, 2012 8:41:02 PM *.*_10037379_jdk6$StreamGobbler run
INFO: 6100 - Message 11
Apr 11, 2012 8:41:02 PM *.*_10037379_jdk6$StreamGobbler run
INFO: Exiting Gobbler for process.295131993.
Apr 11, 2012 8:41:02 PM *.*_10037379_jdk6$StreamGobbler run
INFO: Exiting Gobbler for process.756434719.
Apr 11, 2012 8:41:02 PM *.*_10037379_jdk6$StreamGobbler run
INFO: Exiting Gobbler for process.332711452.
Apr 11, 2012 8:41:02 PM *.*_10037379_jdk6$StreamGobbler run
INFO: Exiting Gobbler for process.1981440623.
Apr 11, 2012 8:41:02 PM *.*_10037379_jdk6$StreamGobbler run
INFO: Exiting Gobbler for process.1043636732.

【讨论】:

  • 哇,感谢您的详尽回答!现在基于此做一个测试实现。真的很感激。
  • 所以我在 Java6 上并且没有重定向。如何防止进程的 stdout/stderr 阻塞?在我的正常用例中,我想写入一个进程并将 stdout/stderr 重定向到单独的日志文件(不阻塞)。
  • @Will 更新为 jdk6 版本。
【解决方案2】:

这里最好的办法是使用 pcntl 函数来分叉一个进程,但是进程之间的通信很困难。我建议创建一个您的进程可以读取的队列,而不是尝试将消息传递到命令行。

Beanstalk 有几个 PHP 客户端,您可以使用它们来处理进程之间的消息传递。

【讨论】:

  • 对不起,也许我的问题不清楚——将编辑。这是一个Java问题。我想要一个长时间运行的cli进程的java线程池(在这种情况下是/usr/bin/php)。我需要能够向池提交内容,然后将其写入 CLI 进程之一的标准输入。