如何使用 rx-java 跟踪文件？

Question

使用rx-java 包，我如何tail 一个文件？

代码：

package tail;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStream;
import java.io.InputStreamReader;
import java.util.logging.Logger;

public class FileUtils {

    public FileUtils() {
    }

    public void readFromInputStream(InputStream inputStream)
            throws IOException {
        StringBuilder resultStringBuilder = new StringBuilder();
        try (BufferedReader br
                = new BufferedReader(new InputStreamReader(inputStream))) {
            String line;
            while ((line = br.readLine()) != null) {
                resultStringBuilder.append(line).append("\n");
            }
        }
        System.out.println(resultStringBuilder);
    }

}

我想在写入文件时解析each line。首先，需要来自 Java 的 equivalent 到 tail -f foo.log。

Answer 1

我不确定具体使用 RxJava。但是您可以采用通用的 Java 解决方案并与 RxJava 结合使用。这当然取决于您希望如何获知。

一个基本选项是使用打开的InputStream（或某些流包装器类或读取器类）来尾随文件。 使用正在运行的线程，您可以使用InputStream.read 等待文件或使用InputStream.available 对新字节数进行采样。在 RxJava 中，您可以使用 IO 调度程序。读完一整行后，只需从线程中调用一些观察者即可。

在核心 Java 中，它可能看起来像：

class ObserveTask implements Runnable {

    private File file;
    private Listener listener;

    @Override
    public void run() {
        try (InputStream stream = new FileInputStream(file); 
            BufferedReader reader = new BufferedReader(stream)) {
            while (!Thread.interrupted()) {
                String line = reader.readLine();
                listener.onNewLine(line);
            }
        }
    }
}

现在只需运行任务：

Scheduler scheduler = Schedulers.newThread();
Scheduler.Worker worker = scheduler.createWorker();
worker.schedule(new ObserveTask());

当然这只是一个基本的例子。

您可以将任务拆分为多个部分，每次运行时都会检查字节，而流保持打开状态。当然这是很危险的，因为这样打开流可能会导致泄漏。

更高级的解决方案可以包括使用WatchService。