java - 跟踪 Gzip 日志文件的 WatcherService

Question

我有一个包含 gzip 压缩日志文件的目录，每行一个事件。为了实时读取和处理这些信息，我创建了一个与此处列出的代码相同的 WatcherService: http://docs.oracle.com/javase/tutorial/essential/io/notification.html

在 processEvents() 方法中，我添加了这段代码来逐行读取已添加或附加的文件:

if (kind == ENTRY_MODIFY) {
    try(BufferedReader reader = new BufferedReader(new InputStreamReader(new GZIPInputStream(Files.newInputStream(child, StandardOpenOption.READ))))) {
        String line;
        while ((line = reader.readLine()) != null) {
            System.out.println(line);
        }
    }
    catch(EOFException ex) {
        //file is empty,  so ignore until next signal
    }
    catch(Exception ex) {
        ex.printStackTrace();
    }
}

现在，正如您所想象的，这对于在几毫秒内创建和关闭的文件非常有效，但是，当处理随时间附加的大文件时，这将一遍又一遍地读取整个文件，每次附加行(假设文件不时被生产者刷新和同步)。

有什么方法可以在每次发送 ENTRY_MODIFY 信号时只读取此文件中的新行，或者找出文件何时“完成”？

如何处理未附加而是覆盖的文件？

Answer 1

首先我想回答您问题的技术方面:

A WatchEvent只为您提供更改(或创建或删除)文件的文件名，仅此而已。因此，如果您需要除此之外的任何逻辑，您必须自己实现(当然也可以使用现有的库)。

如果您只想读取新行，则必须记住每个文件的位置，并且每当该文件发生更改时，您都可以移动到最后一个已知位置。要获得当前位置，您可以使用 CountingInputStream来自 Commons IO 包(学分转到 [1])。要跳转到最后一个位置，可以使用函数 skip .

但是您正在使用 GZIPInputStream，这意味着跳过不会给您带来很大的性能提升，因为跳过压缩流是不可能的。相反，GZIPInputStream skip 将解压缩流，就像您阅读它时一样，因此您只会体验到很小的性能改进(试试吧!)。

我不明白的是您为什么要使用压缩日志文件？你为什么不用 DailyRollingFileAppender 写未压缩的日志呢？并在一天结束时压缩它，当应用程序不再访问它时？

另一种解决方案可能是保留 GZIPInputStream(存储它)，这样您就不必再次重新读取文件。这可能取决于您必须查看多少日志文件来决定这是否合理。

现在有一些关于您的要求的问题:

您没有提到要实时查看日志文件的原因。你为什么不集中你的日志(见 Centralised Java Logging )？例如看看 logstash和此演示文稿(参见 [2] 和 [3])或在 scribe 上或 splunk , 这是商业的(见 [4])。

集中式日志将使您有机会根据您的日志数据真正做出实时 react 。

[1] https://stackoverflow.com/a/240740/734687
[2] Using elasticsearch, logstash & kibana to create realtime dashboards - 幻灯片
[3] Using elasticsearch, logstash & kibana to create realtime dashboards - 视频
[4] Log Aggregation with Splunk - 幻灯片

更新

首先，一个用于生成压缩日志文件的 Groovy 脚本。每次我想模拟日志文件更改时，我都会从 GroovyConsole 启动这个脚本:

// Run with GroovyConsole each time you want new entries
def file = new File('D:\\Projekte\\watcher_service\\data\\log.gz')

// reading previous content since append is not possible
def content
if (file.exists()) {
    def inStream = new java.util.zip.GZIPInputStream(file.newInputStream())
    content = inStream.readLines()
}

// writing previous content and append new data
def random  = new java.util.Random()  
def lineCount = random.nextInt(30) + 1
def outStream = new java.util.zip.GZIPOutputStream(file.newOutputStream())

outStream.withWriter('UTF-8') { writer ->
    if (content) {
        content.each { writer << "$it\n" }
    }
    (1 .. lineCount).each {
        writer.write "Writing line $it/$lineCount\n"
    }
    writer.write '---Finished---\n'
    writer.flush()
    writer.close()
}

println "Wrote ${lineCount + 1} lines."

然后是日志文件阅读器:

import java.nio.file.FileSystems
import java.nio.file.Files
import java.nio.file.Path
import java.nio.file.Paths
import java.nio.file.StandardOpenOption
import java.util.zip.GZIPInputStream
import org.apache.commons.io.input.CountingInputStream
import static java.nio.file.StandardWatchEventKinds.*

class LogReader
{
    private final Path dir = Paths.get('D:\\Projekte\\watcher_service\\data\\')
    private watcher
    private positionMap = [:]
    long lineCount = 0

    static void main(def args)
    {
        new LogReader().processEvents()
    }

    LogReader()
    {
        watcher = FileSystems.getDefault().newWatchService()
        dir.register(watcher, ENTRY_CREATE, ENTRY_DELETE, ENTRY_MODIFY)
    }

    void processEvents()
    {
        def key = watcher.take()
        boolean doLeave = false

        while ((key != null) && (doLeave == false))
        {
            key.pollEvents().each { event ->
                def kind = event.kind()
                Path name = event.context()

                println "Event received $kind: $name"
                if (kind == ENTRY_MODIFY) {
                    // use position from the map, if entry is not there use default value 0
                    processChange(name, positionMap.get(name.toString(), 0))
                }
                else if (kind == ENTRY_CREATE) {
                    processChange(name, 0)
                }
                else {
                    doLeave = true
                    return
                }
            }
            key.reset()
            key = watcher.take()
        }
    }

    private void processChange(Path name, long position)
    {
        // open file and go to last position
        Path absolutePath = dir.resolve(name)
        def countingStream =
                new CountingInputStream(
                new GZIPInputStream(
                Files.newInputStream(absolutePath, StandardOpenOption.READ)))
        position = countingStream.skip(position)
        println "Moving to position $position"

        // processing each new line
        // at the first start all lines are read
        int newLineCount = 0
        countingStream.withReader('UTF-8') { reader ->
            reader.eachLine { line ->
                println "${++lineCount}: $line"
                ++newLineCount
            }
        }
        println "${++lineCount}: $newLineCount new lines +++Finished+++"

        // store new position in map
        positionMap[name.toString()] = countingStream.count
        println "Storing new position $countingStream.count"
        countingStream.close()
    }
}

在函数 processChange 中，您可以看到 1) 输入流的创建。带有 .withReader 的行创建了 InputStreamReader 和 BufferedReader。我总是使用 Grovvy，它是类固醇上的 Java，当你开始使用它时，你就停不下来了。 Java 开发人员应该能够阅读它，但如果您有任何疑问，请发表评论。