如何使用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到tail -f foo.log的equivalent。

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。