我有一个包含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信号时只读取该文件中的新行,或者找出该文件何时“完成”?
如何处理未附加但被覆盖的文件?
发布于 2014-07-14 16:12:31
首先我想回答你问题的技术方面:
WatchEvent只为您提供更改(或创建或删除)文件的文件名,仅此而已。因此,如果您需要任何超出这一点的逻辑,您必须自己实现它(当然,也可以使用现有的库)。
如果您只想读取新的行,您必须记住每个文件的位置,并且无论何时更改该文件,您都可以移动到最后一个已知的位置。要获得当前职位,可以使用来自Commons包的CountingInputStream (学分为1)。要跳转到最后一个位置,可以使用函数skip。
但是您使用的是GZIPInputStream,这意味着跳过不会给您带来很大的性能提升,因为跳过压缩流是不可能的。相反,GZIPInputStream跳过将解压缩流,就像在阅读它时一样,因此您只会体验到很少的性能改进(尝试一下!)。
我不明白的是你为什么要使用压缩的日志文件?为什么不用DailyRollingFileAppender编写未压缩的日志,并在应用程序不再访问它的情况下,在一天结束时压缩它?
另一种解决方案可能是保留GZIPInputStream (存储它),这样您就不必再重新读取文件了。这可能取决于您需要查看多少日志文件才能决定这是否合理。
现在关于您的需求的一些问题:
您没有提到要实时查看日志文件的原因。为什么不集中您的日志(请参阅集中式Java测井)?例如,看看原木存放物和这个演示文稿(参见2和3),或者在文士或扣篮上看一看,这是商业的(参见4)。
集中式日志将使您有机会根据日志数据进行真正的实时响应。
2 -幻灯片
3. -视频
4. -幻灯片
更新
首先是生成压缩日志文件的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开发人员应该能够阅读它,但是如果您有问题,只需评论即可。
https://stackoverflow.com/questions/24555822
复制相似问题