3

我有一个目录，其中包含 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 信号时只读取该文件中的新行，或者找出文件何时“完成”？

如何处理未附加但被覆盖的文件？

4

1 回答 1

5

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

AWatchEvent只是为您提供更改（或创建或删除）文件的文件名，仅此而已。因此，如果您需要除此之外的任何逻辑，您必须自己实现它（或者当然使用现有的库）。

如果您只想读取新行，则必须记住每个文件的位置，并且每当更改此文件时，您都可以移动到最后一个已知位置。要获得当前位置，您可以使用CountingInputStreamCommons IO 包中的 a（学分转到 [1]）。要跳到最后一个位置，可以使用函数skip。

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

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

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

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

您没有提到要实时查看日志文件的原因。为什么不集中您的日志（请参阅Centralized Java Logging）？例如，看看logstash和这个演示文稿（参见 [2] 和 [3]）或scribe或splunk，这是商业的（参见 [4]）。

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

[1] https://stackoverflow.com/a/240740/734687
[2]使用 elasticsearch、logstash 和 kibana 创建实时仪表板- 幻灯片
[3]使用 elasticsearch、logstash 和 kibana 创建实时仪表板- 视频
[4]日志使用 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 开发人员应该能够阅读它，但如果您有问题，请发表评论。

于 2014-07-14T16:12:31.040 回答