从 Java 应用程序启动 Flume 代理

Posted 2023-04-18

【中文标题】从 Java 应用程序启动 Flume 代理

【英文标题】：Starting Flume Agent from Java Application

【发布时间】：2015-02-16 02:58:10

【问题描述】：

我编写了一个 Java 应用程序，它可以构建 Flume 配置文件并将它们写入 Linux 机器上的磁盘。然后应用程序构建flume命令并尝试通过运行以下代码中的命令来启动flume代理：

  try 
                    Process p = Runtime.getRuntime().exec(flumeStartCommand.toString());
                    p.waitFor(); 
                    BufferedReader in = new BufferedReader(new InputStreamReader(p.getInputStream()));
                    String line = null;
                    while ((line = in.readLine()) != null) 
                        System.out.println(line);
                                           

如果我使用 flume 命令并通过终端手动执行它，它可以正常工作。当 java 应用程序尝试执行 flume 命令时，什么也没有发生。这是类路径问题吗？

Flume 命令在这里： /root/flume-flume-1.6/flume-ng-dist/target/apache-flume-1.6.0-SNAPSHOT-bin/apache-flume-1.6.0-SNAPSHOT-bin/bin/flume-ng 代理 --conf conf --conf-file /root/flumeconfs/ConsumerGroup4.conf --name Agent_ConsumerGroup4 -Dflume.root.logger=INFO,console -Dflume.monitoring.type=http -Dflume.monitoring.port=34548

我知道嵌入式代理是另一种选择，但这些仅适用于对我来说不是很有用的 avro sink。欣赏任何想法。 干杯， 科尔曼

【问题讨论】：

你可以把它放在脚本中并在java中调用那个脚本

是的，我也尝试过这样做。创建一个名为 flume.sh 的脚本，第一行启动 bash shell，第二行执行上述查询。仍然没有启动代理，只是挂起......

【参考方案1】：

对于其他试图弄清楚这一点的人来说，flume 命令需要分解为字符串数组中的参数，如下所示：

String[] flumeCommand = new String[]"/root/flume-flume-1.6/flume-ng-dist/target/apache-flume-1.6.0-SNAPSHOT-bin/apache-flume-1.6.0-SNAPSHOT-bin/bin/flume-ng",
                                   "agent",
                                   "--conf",
                                   "conf",
                                   "--conf-file",
                                   _flumeConfigurationDir+Active.getFlumeConfName(),
                                   "--name",
                                   Active.getFlumeAgentName(),
                                   "-Dflume.root.logger=INFO,console",
                                   "-Dflume.monitoring.type=http",
                                   "-Dflume.monitoring.port="+Active.getFlumeMetricsPort(),
;

然后使用以下代码：

ProcessBuilder flumeCommandBuilder = new ProcessBuilder(flumeCommand);
File flumeConfigurationDirFile = new File(_flumeConfigurationDir);
flumeCommandBuilder.directory(flumeConfigurationDirFile);                       
System.out.println("Working Directory: "+ flumeCommandBuilder.directory());
flumeCommandBuilder.redirectErrorStream(true);
Process process = flumeCommandBuilder.start();
BufferedReader reader = new BufferedReader(new InputStreamReader(process.getInputStream()));
String line;
while ((line = reader.readLine()) != null)
    System.out.println("tasklist: " + line);

【讨论】：

@Colman如果我想在远程flume主机而不是本地机器上执行这个命令呢？