2021-11-04 10:07:36
前言
最近我在做一个小工具的时候，有一个要求是可以在Web端实时查看日志文件，相当于在终端执行tail -f命令。没有找到解决这个问题的好办法。一开始我想直接通过FileInputStream读取，因为他也可以跳过N个字节直接读取，就像下面这样。
公共静态void main(字符串[]参数)引发异常(
File File=new File('/home/1 . txt ')；
file inputstream fin=new file inputstream(文件)；
int ch
fin . skip(10)；
while ((ch=fin.read())！=-1){
system . out . print((char)ch)；
}
}
如果不跳过的话，每次都把内容全部看完显示出来显然是不现实的。我们要做的就是每次从最后n行开始读，像tail一样，不断输出最新的行。
另一个问题是能够感知文件的变化，所以我们最终选择直接调用tail命令，通过WebSocket输出到网页。
tail用法
在java中调用tail命令后，它会获取其输入流，并将其包装为BufferedReader。如果无法通过readLine()读取数据，它将始终阻塞，并且不会返回null，这意味着日志文件中暂时没有写入新数据。一旦readLine()方法返回，就意味着新数据已经到达。另一个问题是如何终止它。我们不能让他一直读下去。要在合适的时间终止，答案是当WebSocket断开连接时，Process类提供了一个destroy()方法来终止进程，相当于按Ctrl C。
公共静态void main(字符串[]参数)引发异常(
Process exec=Runtime.getRuntime()。exec(new String[]{'bash '，'-c '，' tail-F/home/HouXinLin/test . txt ' })；
InputStream InputStream=exec . getinputstream()；
buffere reader buffere reader=new buffere reader(new inputStream reader(inputStream))；
for(；){
system . out . println(bufferedreader . readline()' r ')；
}
}
实现过程
有很多方法可以给Spring Boot增加网络套接字功能。目前流行的文章介绍方式有：ServerEndpointExporter、@OnOpen、@OnClose、@OnMessage。这种方式需要声明一个Bean，即ServerEndpointExporter。但是我记得如果你想把它打包成war放入Tomcat运行，需要取消这个Bean，否则会报错，非常麻烦，当然也有办法解决。
还有其他方法可以集成，例如实现WebSocketConfigurer或。
web socket messagebrokerconfigurer接口，但是我目前使用的是web socket messagebrokerconfigurer接口，前端还需要两个库，SockJS和Stomp(这两个库更多的是可选的或者不必要的)。
SockJS提供了类似于WebSocket的对象和一套跨浏览器的API，可以在浏览器和Web服务器之间创建一个低延迟、全双工、跨域的通信通道。如果浏览器不支持WebSocket，也可以模拟对WebSocket的支持。
Stomp是一个简单的面向文本的消息协议，它提供了一种可互操作的连接格式，并允许STOMP客户端与任何STOMP消息代理进行交互。
首先看连接处理层的逻辑，一些不必要的代码就不显示了。
@配置
@EnableWebSocketMessageBroker
公共类WebSocketConfig实现了WebSocketMessageBrokerConfigurer { 0
私有静态最终Logger log=Logger factory . getlogger(websocketconfig . class . getname())；
@自动连线
simpsmessagingtemplate mSimpMessagingTemplate；
@自动连线
WebSocketManager mWebS
ocketManager; @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(WebSocketSession session) throws Exception { log.info("日志监控WebSocket连接,sessionId={}", session.getId()); mWebSocketManager.add(session); super.afterConnectionEstablished(session); } @Override public void afterConnectionClosed(WebSocketSession session, CloseStatus closeStatus) throws Exception { mWebSocketManager.remove(session.getId()); super.afterConnectionClosed(session, closeStatus); } }; } }); } @Override public void registerStompEndpoints(StompEndpointRegistry registry) { registry.addEndpoint("/socket-log") .addInterceptors(new HttpHandshakeInterceptor()) .setHandshakeHandler(new DefaultHandshakeHandler() { @Override protected Principal determineUser(ServerHttpRequest request, WebSocketHandler wsHandler, Map<String, Object> attributes) { return new StompPrincipal(UUID.randomUUID().toString()); } }) .withSockJS(); } @EventListener public void handlerSessionCloseEvent(SessionDisconnectEvent sessionDisconnectEvent) { StompHeaderAccessor headerAccessor = StompHeaderAccessor.wrap(sessionDisconnectEvent.getMessage()); mTailLog.stopMonitor(headerAccessor.getSessionId()); } /** * 路径订阅 * * @param sessionSubscribeEvent */ @EventListener public void handlerSessionSubscribeEvent(SessionSubscribeEvent sessionSubscribeEvent) { StompHeaderAccessor headerAccessor = StompHeaderAccessor.wrap(sessionSubscribeEvent.getMessage()); if (mTailLog.isArriveMaxLog()) { mWebSocketManager.sendMessage(headerAccessor.getSessionId(), "监控数量已经达到限制，无法查看""); log.info("日志监控WebSocket连接已经到达最大数量，将断开sessionId={}", headerAccessor.getSessionId()); mWebSocketManager.close(headerAccessor.getSessionId()); return; } String destination = headerAccessor.getDestination(); String userId = headerAccessor.getUser().getName(); if (destination.startsWith("/user/topic/path")) { String path = destination.substring("/user/topic/path".length()); File file = new File(StringUtils.urlDecoder(path)); if (!file.exists()) { mWebSocketManager.sendMessage(headerAccessor.getSessionId(), "what are you 弄啥嘞，文件找不到啊"); mWebSocketManager.close(headerAccessor.getSessionId()); return; } TailLogListenerImpl tailLogListener = new TailLogListenerImpl(mSimpMessagingTemplate, userId); mTailLog.addMonitor(new LogMonitorObject(file.getName(), file.getParent(), tailLogListener, "" + headerAccessor.getSessionId(), userId)); } } }对于上面的几个接口可能没使用过他的人有点蒙，至少我在学习他的时候是这样的，看上面的代码，我们先要理清逻辑，才能明白为什么要这样写。
实现registerStompEndpoints方法
首先是
WebSocketMessageBrokerConfigurer接口，Spring Boot提供的一个WebSocket配置接口，只需要简简单单地配置两下，就可以实现一个WebSocket程序，这个接口中有8个方法，而我们只需要用到三个个。
然后就是给出前端连接WebSocket所需要的地址，如果连连接地址都不给，后面步骤怎么继续？这个就是通过实现registerStompEndpoints方法来完成，只需要向StompEndpointRegistry中通过addEndpoint添加一个新的”连接点”就可以，还可以设置拦截器，也就是在前端试图连接的时候，如果后端发现这个连接不对劲，有猫腻，可以拒绝和他连接，这步可以通过addInterceptors来完成。
切记如果使用了SocketJs库，那么一定要加入withSockJS。
@Override
public void registerStompEndpoints(StompEndpointRegistry registry) {
registry.addEndpoint("/log")
.addInterceptors(new HttpHandshakeInterceptor())
.setHandshakeHandler(new DefaultHandshakeHandler() {
@Override
protected Principal determineUser(ServerHttpRequest request, WebSocketHandler wsHandler, Map<String, Object> attributes) {
return new StompPrincipal(UUID.randomUUID().toString());
}
})
.withSockJS();
}保存SessionId和WebSocketSession对应关系
这一步是为了方便管理，比如主动断开连接，需要实现
configureWebSocketTransport接口，但是这里的SessionId并不是服务端生成的会话ID，而是这个WebSocket的会话ID，每个WebSocket连接都是不同的。
这里主要考虑到如果前端传过来的文件不存在，那么服务端要能主动断开连接。
@Override
public void configureWebSocketTransport(WebSocketTransportRegistration registration) {
registration.addDecoratorFactory(new WebSocketHandlerDecoratorFactory() {
@Override
public WebSocketHandler decorate(WebSocketHandler webSocketHandler) {
return new WebSocketHandlerDecorator(webSocketHandler) {
@Override
public void afterConnectionEstablished(WebSocketSession session) throws Exception {
log.info("日志监控WebSocket连接,sessionId={}", session.getId());
mWebSocketManager.add(session);
super.afterConnectionEstablished(session);
}
@Override
public void afterConnectionClosed(WebSocketSession session, CloseStatus closeStatus) throws Exception {
mWebSocketManager.remove(session.getId());
super.afterConnectionClosed(session, closeStatus);
}
};
}
});
}监听订阅
接着前端通过Stomp的API来订阅一个消息，那么我们怎么接收订阅的事件呢？就是通过 @EventListener注解来接收SessionSubscribeEvent事件。
而前端订阅时就需要传入要监控的日志路径。这时候我们就能拿到这个WebSocket要监听的日志路径了。
@EventListener
public void handlerSessionSubscribeEvent(SessionSubscribeEvent sessionSubscribeEvent) {
....
}开启tail进程
接着我们要为每个WebSocket都开启一个线程，用来执行tail命令。
@Component
public class TailLog {
public static final int MAX_LOG = 3;
private List<LogMonitorExecute> mLogMonitorExecutes = new CopyOnWriteArrayList<>();
/**
* Log线程池
*/
private ExecutorService mExecutors = Executors.newFixedThreadPool(MAX_LOG);
public void addMonitor(LogMonitorObject object) {
LogMonitorExecute logMonitorExecute = new LogMonitorExecute(object);
mExecutors.execute(logMonitorExecute);
mLogMonitorExecutes.add(logMonitorExecute);
}
public void stopMonitor(String sessionId) {
if (sessionId == null) {
return;
}
for (LogMonitorExecute logMonitorExecute : mLogMonitorExecutes) {
if (sessionId.equals(logMonitorExecute.getLogMonitorObject().getSessionId())) {
logMonitorExecute.stop();
mLogMonitorExecutes.remove(logMonitorExecute);
}
}
}
public boolean isArriveMaxLog() {
return mLogMonitorExecutes.size() == MAX_LOG;
}
}
最终执行者，其中的stop()方法是在WebSocket断开连接时执行的。那么需要事先保存好sessionId和LogMonitorExecute的对应关系。当文件有新变化时，发送给对应的WebSocket。
public class LogMonitorExecute implements Runnable {
private static final Logger log = LoggerFactory.getLogger(LogMonitorExecute.class.getName());
/**
* 监控的对象
*/
private LogMonitorObject mLogMonitorObject;
private volatile boolean isStop = false;
/**
* 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.getSessionId(), 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.close();
} catch (Exception e) {
e.printStackTrace();
}
log.info("{}退出对{}的监控", mLogMonitorObject.getSessionId(), mLogMonitorObject.getPath() + "/" + mLogMonitorObject.getName());
}
public void stop() {
mProcess.destroy();
isStop = true;
}
}
注意这里，要发送给指定的WebSocket，而不是订阅了这个路径的WebSocket，因为使用SimpMessagingTemplate在发送数据时，他可以给所有订阅了此路径的WebSocket，那么就导致如果一个浏览器开了2个监控，而且监控的都是同一个日志文件，那么每个监控都会收到两条同样的消息。
所以要使用convertAndSendToUser方法而不是convertAndSend，这也就是为什么前面会通过setHandshakeHandler设置握手处理器为每个WebSocket连接取一个name的原因。
前端
<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="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="/uploads/allimg/211104/100I63253-0.jpg"></script>
<script src="/lib/stomp/stomp.min.js?x79350"></script>
<script src="/uploads/allimg/211104/100I64933-2.jpg"></script>
<script>
var socket = new SockJS('/socket-log?a=a');
stompClient = Stomp.over(socket);
stompClient.connect({}, function (frame) {
stompClient.subscribe('/user/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.search.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 (false);
}
</script>
</body>
</html>效果
下面是启动、关闭Tomcat的日志。
不通过SimpMessagingTemplate如何发送数据
如果不使用SimpMessagingTemplate，那么首先我们要拿到对应的WebSocketSession，它有个sendMessage方法用来发送数据，但是类型是WebSocketMessage，Spring Boot有几个默认的实现，比如TextMessage用来发送文本信息。
但是如果使用了Stomp，那么单纯地使用他发送是不行的，数据虽然能过去，但是格式不对，Stomp解析不了，所以我们要按照Stomp的格式发送。
但是经过查找，未能找到相关的资料，所以自己看了一下他的源码，其中设计到了StompEncoder这个类，看名字就知道他是Stomp编码的工具。Stomp协议分为三个部分，命令、头、消息体，命令有如下几个：
CONNECT
SEND
SUBSCRIBE
UNSUBSCRIBE
BEGIN
COMMIT
ABORT
ACK
NACK
DISCONNECT紧跟着命令下一行是头，是键值对形式存在的，最后是消息体，末尾以空字符结尾。
下面是发送的必要格式，否则StompEncoder也无法编码，将抛出异常，至于这个为什么这么写，详细就得看
StompEncoderde.writeHeaders方法了，里面有几个验证，这种写完全是被他逼的。
StompEncoder stompEncoder = new StompEncoder();
byte[] encode = stompEncoder.encode(createStompMessageHeader(),msg.getBytes());
webSocketSession.sendMessage(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.txt
hxl 1368 29021 0 09:02 pts/0 00:00:00 grep 1.txt
hxl 20298 29672 0 09:00 pts/6 00:00:00 tail -f 1.txt
hxl@hxl-PC:/home/HouXinLin/test/tail$ ls -l /proc/20298/fd
总用量 0
lrwx------ 1 hxl hxl 64 3月 16 09:02 0 -> /dev/pts/6
lrwx------ 1 hxl hxl 64 3月 16 09:02 1 -> /dev/pts/6
lrwx------ 1 hxl hxl 64 3月 16 09:02 2 -> /dev/pts/6
lr-x------ 1 hxl hxl 64 3月 16 09:02 3 -> /home/HouXinLin/test/tail/2.txt
lr-x------ 1 hxl hxl 64 3月 16 09:02 4 -> anon_inode:inotify
hxl@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
总用量 0
lrwx------ 1 hxl hxl 64 3月 16 09:02 0 -> /dev/pts/6
lrwx------ 1 hxl hxl 64 3月 16 09:02 1 -> /dev/pts/6
lrwx------ 1 hxl hxl 64 3月 16 09:02 2 -> /dev/pts/6
lr-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:inotify
hxl@hxl-PC:/home/HouXinLin/test/tail$
下一篇：没有了
