前言

最近在做一个小工具，有个需求是在web端能实时查看日志文件，也就是相当于在终端执行tail -f命令，对此没有找到好的解决方式，一开始想得直接通过fileinputstream来读取，因为他也能直接跳过n个字节来读取，就像下面这样。

public static void main(string[] args) throws exception {    file file = new file("/home/1.txt");    fileinputstream fin = new fileinputstream(file);    int ch;    fin.skip(10);    while ((ch = fin.read()) != -1){      system.out.print((char) ch);    }  }

如果不跳过的话，那么每次读取全部内容并展示显然不现实，我们要做的是像tail一样，每次从后n行开始读取，并且会持续输出最新的行。

还有一个问题就是对文件的变化要能感知到，所以最后选择直接调用tail命令，并且通过websocket输出到网页上。

tail用法

在java中调用tail命令后，拿到它的输入流并且包装成bufferedreader，如果通过readline()读取不到数据，那么他会一直阻塞，并不会返回null，这也就代表日志文件中暂时还没有新数据写入，一旦readline()方法返回，那么就代表有新数据到达了。另外一个问题就是如何终止，我们不可能让他一直读取，要在一个合适的时间终止，答案就是在websocket断开连接时，并且process类提供了destroy()方法用来终止这个进程，相当于按下了ctrl+c

 public static void main(string[] args) throws exception {     process exec = runtime.getruntime().exec(new string[]{"bash", "-c", "tail -f /home/houxinlin/test.txt"});     inputstream inputstream = exec.getinputstream();     bufferedreader bufferedreader = new bufferedreader(new inputstreamreader(inputstream));     for (;;){         system.out.println(bufferedreader.readline()+"r");     } }

实现过程

在spring boot中加入websocket功能有很多方式，目前感觉普遍的文章都是介绍以rverendpointexporter、@onopen、 @onclo、@onmessage这种方式来实现的，这种方式需要声明一个bean，也就是rverendpointexporter，但是我记得如果要打包成war放入tomcat中运行时，还需要把这个bean取消掉，否则还会报错，非常的麻烦，当然也有办法解决。

还有其他集成的办法，比如实现websocketconfigurer或者
websocketmessagebrokerconfigurer接口，而我目前采用的是实现websocketmessagebrokerconfigurer接口，并且前端还需要两个库，sockjs和stomp（更具选择，也可以不使用）。

sockjs提供类似于websocket的对象，还有一套跨浏览器的api，可以在浏览器和web服务器之间创建了低延迟，全双工，跨域的通信通道，如果浏览器不支持 websocket，它还可以模拟对websocket的支持。

stomp即simple text orientated messaging protocol，简单(流)文本定向消息协议，它提供了一个可互操作的连接格式，允许stomp客户端与任意stomp消息代理（broker）进行交互。

首先看一下连接处理层的逻辑，其中一部分非必要的代码就不展示了。

@configuration@enablewebsocketmessagebrokerpublic class websocketconfig implements websocketmessagebrokerconfigurer {    private static final logger log = loggerfactory.getlogger(websocketconfig.class.getname());    @autowired    simpmessagingtemplate msimpmessagingtemplate;    @autowired    websocketmanager mwebsocketmanager;    @autowired    taillog mtaillog;    @override    public void configuremessagebroker(messagebrokerregistry registry) {        registry.enablesimplebroker("/topic/path");    }    @override    public void configurewebsockettransport(websockettransportregistration registration) {        registration.adddecoratorfactory(new websockethandlerdecoratorfactory() {            @override            public websockethandler decorate(websockethandler websockethandler) {                return new websockethandlerdecorator(websockethandler) {                    @override                    public void afterconnectionestablished(websocketssion ssion) throws exception {                        log.info("日志监控websocket连接,ssionid={}", ssion.getid()); 向量a                       mwebsocketmanager.add(ssion);                        super.afterconnectionestablished(ssion);                    }                    @override                    public void afterconnectionclod(websocketssion ssion, clostatus clostatus) throws exception {             代言英语           mwebsocketmanager.remove(ssion.getid());                        super.afterconnectionclod(ssion, clostatus);                    }                };            }        });    }    @override    public void registerstompendpoints(stompendpointregistry registry) {        registry.addendpoint("/socket-log")                .addinterceptors(new httphandshakeinterceptor())                .thandshakehandler(new defaulthandshakehandler() {                    @override                    protected principal determineur(rverhttprequest request, websockethandler wshandler, map<string, object> attributes) {                        return new stompprincipal(uuid.randomuuid().tostring());                    }                })                .withsockjs();    }    @eventlistener    public void handlerssioncloevent(ssiondisconnectevent ssiondisconnectevent) {        stompheaderaccessor headeraccessor = stompheaderaccessor.wrap(ssiondisconnectevent.getmessage());        mtaillog.stopmonitor(headeraccessor.getssionid());    }    /**     * 路径订阅     *     * @param ssionsubscribeevent     */    @eventlistener    public void handlerssionsubscribeevent(ssionsubscribeevent ssionsubscribeevent) {        stompheaderaccessor headeraccessor = stompheaderaccessor.wrap(ssionsubscribeevent.getmessage());        if (mtaillog.isarrivemaxlog()) {            mwebsocketmanager.ndmessage(headeraccessor.getssionid(), "监控数量已经达到限制，无法查看"");            log.info("日志监控websocket连接已经到达最大数量，将断开ssionid={}", headeraccessor.getssionid());            mwebsocketmanager.clo(headeraccessor.getssionid());            return;        }        string destination = headeraccessor.getdestination();        string urid = headeraccessor.getur().getname();        if (destination.startswith("/ur/topic/path")) {            string path = destination.substring("/ur/topic/path".length());            file file = new file(stringutils.urldecoder(path));            if (!file.exists()) {                mwebsocketmanager.ndmessage(headeraccessor.getssionid(), "what are you 弄啥嘞，文件找不到啊");                mwebsocketmanager.clo(headeraccessor.getssionid());                return;            }            tailloglistenerimpl tailloglistener = new tailloglistenerimpl(msimpmessagingtemplate, urid);            mtaillog.addmonitor(new logmonitorobject(file.getname(), file.getparent(),                    tailloglistener, "" + headeraccessor.getssionid(), urid));        }    }}

对于上面的几个接口可能没使用过他的人有点蒙，至少我在学习他的时候是这样的，看上面的代码，我们先要理清逻辑，才能明白为什么要这样写。

实现registerstompendpoints方法

首先是
websocketmessagebrokerconfigurer接口，spring boot提供的一个websocket配置接口，只需要简简单单地配置两下，就可以实现一个websocket程序，这个接口中有8个方法，而我们只需要用到三个个。

然后就是给出前端连接websocket所需要的地址，如果连连接地址都不给，后面步骤怎么继续？这个就是通过实现registerstompendpoints方法来完成，只需要向stompendpointregistry中通过addendpoint添加一个新的”连接点”就可以，还可以设置拦截器，也就是在前端试图连接的时候，如果后端发现这个连接不对劲，有猫腻，可以拒绝和他连接，这步可以通过addinterceptors来完成。

切记如果使用了socketjs库，那么一定要加入withsockjs。

@overridepublic void registerstompendpoints(stompendpointregistry registry) {    registry.addendpoint("/log")            .addinterceptors(new httphandshakeinterceptor())            .thandshakehandler(new defaulthandshakehandler() {                @override                protected principal determineur(rverhttprequest request, websockethandler wshandler, map<string, object> attributes) {                    return new stompprincipal(uuid.randomuuid().tostring());                }            })            .withsockjs();}

保存ssionid和websocketssion对应关系

这一步是为了方便管理，比如主动断开连接，需要实现
configurewebsockettransport接口，但是这里的ssionid并不是服务端生成的会话id，而是这个websocket的会话id，每个websocket连接都是不同的。

这里主要考虑到如果前端传过来的文件不存在，那么服务端要能主动断开连接。

@overridepublic void configurewebsockettransport(websockettransportregistration registration) {    registration.adddecoratorfactory(new websockethandlerdecoratorfactory() {        @override        public websockethandler decorate(websockethandler websockethandler) {            return new websockethandlerdecorator(websockethandler) {                @override                public void afterconnectionestablished(websocketssion ssion) throws exception {                    log.info("日志监控websocket连接,ssionid={}", ssion.getid());                    mwebsocketmanager.add(ssion);                    super.afterconnectionestablished(ssion);                }                @override                public void afterconnectionclod(websocketssion ssion, clostatus clostatus) throws exception {                    mwebsocketmanager.remove(ssion.getid());                    super.afterconnectionclod(ssion, clostatus);                }            };        }    });}

监听订阅

接着前端通过stomp的api来订阅一个消息，那么我们怎么接收订阅的事件呢？就是通过 @eventlistener注解来接收ssionsubscribeevent事件。

而前端订阅时就需要传入要监控的日志路径。这时候我们就能拿到这个websocket要监听的日志路径了。

@eventlistenerpublic void handlerssionsubscribeevent(ssionsubscribeevent ssionsubscribeevent) {    ....}

开启tail进程

接着我们要为每个websocket都开启一个线程，用来执行tail命令。

@componentpublic class taillog {    public static final int max_log = 3;    private list<logmonitorexecute> mlogmonitorexecutes = new copyonwritearraylist<>();    /**     * log线程池     */    private executorrvice mexecutors = executors.newfixedthreadpool(max_log);    public void addmonitor(logmonito集合的关系robject object) {        logmonitorexecute logmonitorexecute = new logmonitorexecute(object);        mexecutors.execute(logmonitorexecute);        mlogmonitorexecutes.add(logmonitorexecute);    }    public void stopmonitor(string ssionid) {        if (ssionid == null) {            return;        }        for (logmonitorexecute logmonitorexecute : mlogmonitorexecutes) {            if (ssionid.equals(logmonitorexecute.getlogmonitorobject().getssionid())) {                logmonitorexecute.stop();                mlogmonitorexecutes.remove(logmonitorexecute);            }        }    }    public boolean isarrivemaxlog() {        return mlogmonitorexecutes.size() == max_log;    }}

最终执行者，其中的stop()方法是在websocket断开连接时执行的。那么需要事先保存好ssionid和logmonitorexecute的对应关系。当文件有新变化时，发送给对应的websocket。

public class logmonitorexecute implements runnable {    private static final logger log = loggerfactory.getlogger(logmonitorexecute.class.getname());    /**     * 监控的对象     */    private logmonitorobject mlogmonitorobject;    private volatile boolean isstop = fal;    /**     * tail 进程对象     */    private process mprocess;    public logmonitorexecute(logmonitorobject logmonitorobject) {        mlogmonitorobject = logmonitorobject;    }    public logmonitorobject getlogmonitorobject() {        return mlogmonitorobject;    }    @override    public void run() {        try {            string path = paths.get(mlogmonitorobject.getpath(), mlogmonitorobject.getname()).tostring();            log.info("{}对{}开始进行日志监控", mlogmonitorobject.getssionid(), path);            mprocess = runtime.getruntime().exec(new string[]{"bash", "-c", "tail -f " + path});            inputstream inputstream = mprocess.getinputstream();            bufferedreader mbufferedreader = new bufferedreader(new inputstreamreader(inputstream, "utf-8"));            string buffer = null;            while (!thread.currentthread().isinterrupted() && !isstop) {                buffer = mbufferedreader.readline();                if (mlogmonitorobject.gettailloglistener() != null) {                    mlogmonitorobject.gettailloglistener().onnewline(mlogmonitorobject.getname(), mlogmonitorobject.getpath(), buffer);                    continue;               本兮怎么死的 }                break;            }            mbufferedreader.clo();        } catch (exception e) {            e.printstacktrace();        }        log.info("{}退出对{}的监控", mlogmonitorobject.getssionid(), mlogmonitorobject.getpath() + "/" + mlogmonitorobject.getname());    }    public void stop() {        mprocess.destroy();        isstop = true;    }}

注意这里，要发送给指定的websocket，而不是订阅了这个路径的websocket，因为使用simpmessagingtemplate在发送数据时，他可以给所有订阅了此路径的websocket，那么就导致如果一个浏览器开了2个监控，而且监控的都是同一个日志文件，那么每个监控都会收到两条同样的消息。

所以要使用convertandndtour方法而不是convertandnd，这也就是为什么前面会通过thandshakehandler设置握手处理器为每个websocket连接取一个name的原因。

前端

<!doctype html><html lang="en"><head>    <meta chart="utf-8">    <title>日志监控</title>    <style>        body {            background: #000000;            color: #ffffff;        }        .log-list {            color: #ffffff;            font-size: 13px;            padding: 25px;        }    </style></head><body><div class="container">    <div class="log-list">    </div></div><script src="https://cdn.jsdelivr.net/npm/sockjs-client@1/dist/sockjs.min.js"></script><script src="/lib/stomp/stomp.min.js?x32330"></script><script src="https://lib.sinaapp.com/js/jquery/2.0.2/jquery-2.0.2.min.js"></script><script>    var socket = new sockjs('/socket-log?a=a');    stompclient = stomp.over(socket);    stompclient.connect({}, function (frame) {        stompclient.subscribe('/ur/topic/path'+getqueryvariable("path"), function (greeting) {            console.log("a" + greeting)            let item = $("<div class='log-line'></div>");            item.text(greeting.body)            $(".log-list").append(item);            $("html, body").animate({scrolltop: $(document).height()}, 0);        });    });    function getqueryvariable(variable) {     皇帝的新装教学设计   var query = window.location.arch.substring(1);        var vars = query.split("&");        for (var i = 0; i < vars.length; i++) {            var pair = vars[i].split("=");            if (pair[0] == variable) {                return encodeuricomponent(pair[1]);            }        }        return (fal);    }</script></body></html>

效果

下面是启动、关闭tomcat的日志。

不通过simpmessagingtemplate如何发送数据

如果不使用simpmessagingtemplate，那么首先我们要拿到对应的websocketssion，它有个ndmessage方法用来发送数据，但是类型是websocketmessage，spring boot有几个默认的实现，比如textmessage用来发送文本信息。

但是如果使用了stomp，那么单纯地使用他发送是不行的，数据虽然能过去，但是格式不对，stomp解析不了，所以我们要按照stomp的格式发送。

但是经过查找，未能找到相关的资料，所以自己看了一下他的源码，其中设计到了stompencoder这个类，看名字就知道他是stomp编码的工具。stomp协议分为三个部分，命令、头、消息体，命令有如下几个：

connectndsubscribeunsubscribebegincommitabortacknackdisconnect

紧跟着命令下一行是头，是键值对形式存在的，最后是消息体，末尾以空字符结尾。

下面是发送的必要格式，否则stompencoder也无法编码，将抛出异常，至于这个为什么这么写，详细就得看
stompencoderde.writeheaders方法了，里面有几个验证，这种写完全是被他逼的。

 stompencoder stompencoder = new stompencoder(); byte[] encode = stompencoder.encode(createstompmessageheader(),msg.getbytes()); websocketssion.ndmessage(new textmessage(encode));  private hashmap<string, object> createstompmessageheader() {     hashmap<string, object> hashmap = new hashmap<>();     hashmap.put("subscription", createlist("sub-0"));     hashmap.put("content-type", createlist("text/plain"));     hashmap<string, object> stringobjecthashmap = new hashmap<>();     stringobjecthashmap.put("simpmessagetype", simpmessagetype.message);     stringobjecthashmap.put("stompcommand", stompcommand.message);     stringobjecthashmap.put("subscription", "sub-0");     stringobjecthashmap.put("nativeheaders", hashmap);     return stringobjecthashmap;} private list<string> createlist(string value) {    list<string> list = new arraylist<>();    list.add(value);    return list;}

tail -f 为什么会失效

这是偶尔间的一个发现，当执行tail -f命令后，我们通过vim、gedit等工具编辑并保存这个文件，会发现tail -f并不会输出新的行，反而通过echo test>>xx.txt是正常的。

那这里的蹊跷又在哪？

其实，tail -f不管在文件移动、改名都会进行追踪，因为他跟踪的是文件描述符，引入维基百科的一句话：

文件描述符在形式上是一个非负整数。实际上，它是一个索引值，指向内核为每一个进程所维护的该进程打开文件的记录表。当程序打开一个现有文件或者创建一个新文件时，内核向进程返回一个文件描述符。在程序设计中，一些涉及底层的程序编写往往会围绕着文件描述符展开。但是文件描述符这一概念往往只适用于unix、linux这样的操作系统。

tail -f执行后会产生一个进程，可以在/proc/pid/fd路径下查看他所打开的文件描述符，下面来看一个gif。

在这个操作中，首先在终端1中创建一个1.txt，然后进行tail -f跟踪，接着在终端2中追加一行数据，可以看到终端1中是可以打印出来的。

然后再看神奇的一幕，在终端2进行mv改名，接着向被改名后的文件追加新的一行，你会发现，终端1居然还是会打印的。

如果查看一下这个进程的文件描述符，就不为奇了，在下面的命令中，显示了3号描述符追踪的是
/home/houxinlin/test/tail/2.txt。

hxl@hxl-pc:/home/houxinlin/test/tail$ ps -ef |grep 1.txthxl       1368 29021  0 09:02 pts/0    00:00:00 grep 1.txthxl      20298 29672  0 09:00 pts/6    00:00:00 tail -f 1.txthxl@hxl-pc:/home/houxinlin/test/tail$ ls -l /proc/20298/fd总用量 0lrwx------ 1 hxl hxl 64 3月  16 09:02 0 -> /dev/pts/6lrwx------ 1 hxl hxl 64 3月  16 09:02 1 -> /dev/pts/6lrwx------ 1 hxl hxl 64 3月  16 09:02 2 -> /dev/pts/6lr-x------ 1 hxl hxl 64 3月  16 09:02 3 -> /home/houxinlin/test/tail/2.txtlr-x------ 1 hxl hxl 64 3月  16 09:02 4 -> anon_inode:inotifyhxl@hxl-pc:/home/houxinlin/test/tail$ 

但是如果我们通过vim、等工具编辑这个文件后，那么这个文件描述符中会被记录为被删除，即使这个文件确实是存在的，此时在向2.txt文件中追加就会失效。

hxl@hxl-pc:/home/houxinlin/test/tail$ vim 2.txt hxl@hxl-pc:/home/houxinlin/test/tail$ ls -l /proc/20298/fd总用量 0lrwx------ 1 hxl hxl 64 3月  16 09:02 0 -> /dev/pts/6lrwx------ 1 hxl hxl 64 3月  16 09:02 1 -> /dev/pts/6lrwx------ 1 hxl hxl 64 3月  16 09:02 2 -> /dev/pts/6lr-x------ 1 hxl hxl 64 3月  16 09:02 3 -> /home/houxinlin/test/tail/2.txt~ (deleted)lr-x------ 1 hxl hxl 64 3月  16 09:02 4 -> anon_inode:inotifyhxl@hxl-pc:/home/houxinlin/test/tail$