1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130

@Service
public class LogMonitorServiceImpl implements LogMonitorService {

    private final Logger log = LoggerFactory.getLogger(LogMonitorServiceImpl.class);

    /**
     * 日志根目录、默认目录
     */
    @Value("${log.monitor.defaultPath}")
    private String logRootPath;

    /**
     * 获取路径
     *
     * @param logPath
     * @return
     */
    private String getLogPath(String logPath) throws CommonException {
        String path = null;
        if (ObjectUtils.isEmpty(logPath)) {
            path = logRootPath;
        } else {
            if (!logPath.contains(logRootPath)) {
                throw new CommonException(ResponseEnum.NO_ACCESS_FOR_THIS_PATH);
            }
            path = logPath;
        }
        return path;
    }

    /**
     * 按照时间查询日志
     *
     * @param startTime
     * @param endTime
     * @param dir
     * @return
     */
    private List<String> queryLogByTime(Long startTime, Long endTime, File dir) {
        List<String> files = new LinkedList<>();
        for (String s : Objects.requireNonNull(dir.list())) {
            File file = new File(dir, s);
            long lastModified = file.lastModified();
            if (startTime <= lastModified && endTime >= lastModified) {
                files.add(file.getAbsolutePath().replaceAll("\\\\", "/"));
            }
        }
        log.info("queryLogByTime");
        return files;
    }

    /**
     * 查询所有日志
     *
     * @param dir
     * @return
     */
    private List<String> queryLogWithoutTime(File dir) {
        List<String> files = new LinkedList<>();
        for (String s : Objects.requireNonNull(dir.list())) {
            File file = new File(dir, s);
            files.add(file.getAbsolutePath().replaceAll("\\\\", "/"));
        }
        log.info("queryLogWithoutTime");
        return files;
    }


    @Override
    public List<String> queryLog(QueryLogRequest request) throws CommonException {
        String logPath = getLogPath(request.getLogPath());
        File dir = new File(logPath);
        if (!dir.exists() || !dir.isDirectory()) {
            throw new CommonException(ResponseEnum.DIR_NOT_EXIST_OR_DIR_IS_A_FILE);
        }
        if (!ObjectUtils.isEmpty(request.getStartTime()) && !ObjectUtils.isEmpty(request.getEndTime())) {
            return queryLogByTime(request.getStartTime(), request.getEndTime(), dir);
        } else {
            return queryLogWithoutTime(dir);
        }
    }

    @Override
    public String downloadLog(DownloadLogRequest request, HttpServletResponse response) throws CommonException {
        String logPath = getLogPath(request.getLogPath());
        File file = new File(logPath);
        if (!file.exists() || !file.isFile()) {
            throw new CommonException(ResponseEnum.FILE_NOT_EXIST_OR_FILE_IS_DIRECTORY);
        }
        // 实现文件下载
        byte[] buffer = new byte[1024];
        FileInputStream fis = null;
        BufferedInputStream bis = null;
        try {
            // 配置文件下载
            response.setHeader("content-type", "application/octet-stream");
            response.setContentType("application/octet-stream");
            // 下载文件能正常显示中文
            response.setHeader("Content-Disposition", "attachment;filename=" + URLEncoder.encode(request.getDownloadName(), "UTF-8"));
            fis = new FileInputStream(file);
            bis = new BufferedInputStream(fis);
            OutputStream os = response.getOutputStream();
            int i = bis.read(buffer);
            while (i != -1) {
                os.write(buffer, 0, i);
                i = bis.read(buffer);
            }
            log.info("下载文件成功!");
        } catch (Exception e) {
            log.info("下载文件失败!");
            throw new CommonException(ResponseEnum.DOWNLOAD_FILE_FAILED);
        } finally {
            if (bis != null) {
                try {
                    bis.close();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
            if (fis != null) {
                try {
                    fis.close();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
        }
        return null;
    }
}

1
2
3
4

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-websocket</artifactId>
</dependency>

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12

@Configuration
@EnableWebSocket
public class WebSocketConfig implements WebSocketConfigurer {

    @Value("${log.websocket}")
    private String path;

    @Override
    public void registerWebSocketHandlers(WebSocketHandlerRegistry registry) {
        registry.addHandler(new SocketEventHandler(), path).setAllowedOrigins("*");
    }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17

public class SocketEventHandler extends AbstractWebSocketHandler {

    private final Logger log = LoggerFactory.getLogger(SocketEventHandler.class);

    @Override
    public void afterConnectionEstablished(WebSocketSession session) {
        log.info("进行连接");
        WebSocketUtils.addSessoin(session);
        WebSocketUtils.startMonitor(session.getId());
    }

    @Override
    public void afterConnectionClosed(WebSocketSession session, CloseStatus status) {
        log.info("关闭连接");
        WebSocketUtils.reduceSession(session);
    }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84

public class WebSocketUtils {

    private static final Logger log = LoggerFactory.getLogger(WebSocketUtils.class);
    /**
     * 已连接的websocket
     */
    private static Map<String, WebSocketSession> onlineSession = new HashMap<>();

    /**
     * 添加用户
     *
     * @param session
     */
    public static void addSessoin(WebSocketSession session) {
        onlineSession.put(session.getId(), session);
        log.info("{}的用户连接websocket", session.getId());
    }

    /**
     * 移除用户
     *
     * @param session
     */
    public static void reduceSession(WebSocketSession session) {
        onlineSession.remove(session.getId());
        log.info("{}的用户断开websocket", session.getId());
    }

    /**
     * 开启监测
     * 本质是一监控一线程
     *
     * @param sessionId
     */
    public static void startMonitor(String sessionId) {
        WebSocketSession session = onlineSession.get(sessionId);
        String query = session.getUri().getQuery();
        String logPath = query.substring(query.indexOf("=") + 1);
        new FileMonitor(session.getId(), logPath);
    }

    /**
     * 关闭监控
     * session关闭，相应线程也会关闭
     *
     * @param sessionId
     */
    public static void endMonitor(String sessionId) {
        WebSocketSession session = onlineSession.get(sessionId);
        sendMessageTo(sessionId,"<error>ERROR 监控线程出现异常！</error>");
        try {
            session.close();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    /**
     * 发送消息给指定用户
     *
     * @param sessionId
     * @param message
     */
    public static void sendMessageTo(String sessionId, String message) {
        WebSocketSession session = onlineSession.get(sessionId);
        try {
            session.sendMessage(new TextMessage(message));
        } catch (Exception e) {
            e.printStackTrace();
            WebSocketUtils.endMonitor(sessionId);
        }
    }

    /**
     * session是否在线
     * 用于决定线程是否关闭
     *
     * @param sessionId
     * @return
     */
    public static boolean currentSessionAlive(String sessionId) {
        return onlineSession.containsKey(sessionId);
    }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38

public class FileMonitor {

    private static final Logger log = LoggerFactory.getLogger(FileMonitor.class);

    /**
     * 绑定的websocket
     */
    private String sessionId;

    /**
     * 绑定的监控日志路径
     */
    private String logPath;

    /**
     * 监控时间间隔，单位ms
     */
    private Long monitorDelay;

    public FileMonitor(String sessionId, String logPath) {
        this.sessionId = sessionId;
        this.logPath = logPath;
        this.monitorDelay = 500L;
        startFileMonitor(monitorDelay);
    }

    public FileMonitor(String sessionId, String logPath, Long monitorDelay) {
        this.sessionId = sessionId;
        this.logPath = logPath;
        this.monitorDelay = monitorDelay;
        startFileMonitor(monitorDelay);
    }

    private void startFileMonitor(Long monitorDelay) {
        Thread thread = new Thread(new FileMonitorRunnable(sessionId, logPath, monitorDelay));
        thread.start();
    }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78

public class FileMonitorRunnable implements Runnable {

    private static final Logger log = LoggerFactory.getLogger(FileMonitorRunnable.class);

    private ByteBuffer byteBuffer = ByteBuffer.allocate(1024 * 100);

    private CharBuffer charBuffer = CharBuffer.allocate(1024 * 50);

    private CharsetDecoder decoder = StandardCharsets.UTF_8.newDecoder();

    private boolean isRunning = true;

    private String sessionId;

    private String logPath;

    private Long monitorDelay;

    public FileMonitorRunnable(String sessionId, String logPath, Long monitorDelay) {
        this.sessionId = sessionId;
        this.logPath = logPath;
        this.monitorDelay = monitorDelay;
    }

    @Override
    public void run() {
        File file = new File(logPath);
        FileChannel channel = null;
        try {
            channel = new FileInputStream(file).getChannel();
            channel.position(channel.size());
        } catch (Exception e) {
            log.info("监控文件失败，检查路径是否正确");
            WebSocketUtils.endMonitor(sessionId);
            e.printStackTrace();
        }
        long lastModified = file.lastModified();
        //TODO: 初次连接将所有内容丢回去？这个考虑到数据如果很多先不丢
        while (isRunning) {
            long now = file.lastModified();
//            log.info("{}的连接正在通过线程{}监控{}文件",sessionId,Thread.currentThread().getName(),logPath);
            if (now != lastModified) {
                log.info("{}的连接正在通过线程{}监控{}的文件update", sessionId, Thread.currentThread().getName(), logPath);
                String newContent = getNewContent(channel);
                WebSocketUtils.sendMessageTo(sessionId, newContent);
                lastModified = now;
            }
            try {
                Thread.sleep(monitorDelay);
            } catch (InterruptedException e) {
                e.printStackTrace();
                WebSocketUtils.endMonitor(sessionId);
            }
            isRunning = WebSocketUtils.currentSessionAlive(sessionId);

        }
    }

    private String getNewContent(FileChannel channel) {
        try {
            byteBuffer.clear();
            charBuffer.clear();
            int length = channel.read(byteBuffer);
            if (length != -1) {
                byteBuffer.flip();
                decoder.decode(byteBuffer, charBuffer, true);
                charBuffer.flip();
                return charBuffer.toString();
            } else {
                channel.position(channel.size());
            }
        } catch (Exception e) {
            e.printStackTrace();
            WebSocketUtils.endMonitor(sessionId);
        }
        return null;
    }
}