Spring Boot 集成 WebSocket 实例 | 前端持续打印远程日志文件更新内容（模拟 tail 命令）

news2024/11/17 3:30:23

这个是我在 CSDN 的第一百篇原则博文，留念😎

#1 需求说明

先说下项目结构，后端基于 Spring Boot 3，前端为 node.js 开发的控制台程序。现在希望能够在前端模拟 tail 命令，持续输出后端的日志文件。

#2 技术方案

#2.1 基于轮询（PASS）

这个方案实施较为简单，通过前端不断（定时）发起请求，并携带已读的内容坐标（position），询问后端日志文件是否有更新，判断依据为当前文件大小大于 position。若有变动，则读取更新的内容，回显在前端控制台。

此方案会产生非常多的请求，如果定时间隔设置不好，会有明显的延迟，故不采用。

#2.2 WebSocket 长连接

前端开启一个 WebSocket
后端监听到长连接后，启动文件变动检测线程
若文件发生变动，则读取更新内容，发送到前端

#3 实施

#3.1 后端改造

关于 Spring Boot 与 WebSocket 的集成，请转到：springboot集成websocket持久连接（权限过滤+拦截）

首先，我们定义一个监听文件变动并读取最新内容的工具类（借助于 common-io 包）：

class FileTail(val path:Path, val handler: Consumer<String>, delay:Long=1000): FileAlterationListenerAdaptor() {
    private val watcher = FileSystems.getDefault().newWatchService()

    private val MODE = "r"
    private var reader = RandomAccessFile(path.toFile(), MODE)
    private var position= reader.length()

    // 使用 JDK 自带的 WatchService ，发现不能正常读取文件追加的内容
    private var monitor: FileAlterationMonitor = FileAlterationMonitor(delay)

    init {
        // 初始化监视器，只检测同名的文件
        FileAlterationObserver(path.parent.toFile()) { f: File -> f.name == path.name }.also { observer->
            observer.addListener(this)
            monitor.addObserver(observer)

            monitor.start()
        }
    }

    override fun onFileChange(file: File) {
        reader.seek(position)

        val bytes = mutableListOf<Byte>()
        val tmp = ByteArray(1024)
        var readSize: Int

        while ((reader.read(tmp).also { readSize = it }) != -1) {
            for (i in 0..< readSize){
                bytes.add(tmp[i])
            }
        }

        position += bytes.size
        handler.accept(String(bytes.toByteArray()))
    }
    
    fun stop() {
        reader.close()
        monitor.stop()
    }
}

再定义长连接的通信处理类：

@Component
class FileTailWsHandler : TextWebSocketHandler() {
    private val logger = LoggerFactory.getLogger(javaClass)

    companion object {
        val monitors = mutableMapOf<String, FileTail>()
    }

    override fun afterConnectionEstablished(session: WebSocketSession) {
        try{
             val textFile = Paths.get("logs/spring.log")

            // 加入队列
            monitors[session.id] = FileTail(
                textFile,
                { text -> session.sendMessage(TextMessage(text)) }
            )
        }catch (e:Exception){
            logger.error("处理客户端消息失败", e)
            session.sendMessage(TextMessage("服务器出错：${ExceptionUtils.getMessage(e)}"))
            session.close(CloseStatus.SERVER_ERROR)
        }
    }

    override fun afterConnectionClosed(session: WebSocketSession, status: CloseStatus) {
        logger.info("客户端（${session.id}）${session.remoteAddress} 断开连接...")

        monitors.remove(session.id)?.stop()
    }
}

编写配置类，启用上述的组件：

@Component
class WsInterceptor : HandshakeInterceptor {
    private val logger = LoggerFactory.getLogger(javaClass)

    override fun beforeHandshake(
        request: ServerHttpRequest,
        response: ServerHttpResponse,
        wsHandler: WebSocketHandler,
        attributes: MutableMap<String, Any>
    ): Boolean {
        if(logger.isDebugEnabled){
            logger.debug("WS 握手开始：${request.uri} 客户端=${request.remoteAddress}")
            request.headers.forEach { name, v -> logger.debug("[HEADER] $name = $v") }
        }
		//此处可以进行鉴权
	
		//写入属性值，方便在 handler 中获取
        attributes[F.PARAMS]    = request.headers.getFirst(F.PARAMS)?: EMPTY
        // 返回 true 才能建立连接
        return true
    }

    override fun afterHandshake(
        request: ServerHttpRequest,
        response: ServerHttpResponse,
        wsHandler: WebSocketHandler,
        exception: Exception?
    ) {
    }
}

@Configuration
@EnableWebSocket
class SocketConfig : WebSocketConfigurer {
    private val logger = LoggerFactory.getLogger(javaClass)

    @Resource
    lateinit var interceptor: WsInterceptor
    @Resource
    lateinit var fileTailHandler:FileTailWsHandler

    override fun registerWebSocketHandlers(registry: WebSocketHandlerRegistry) {
        registry.addHandler(fileTailHandler, "/ws/file-tail").addInterceptors(interceptor)
	}
}

#3.2 前端（node.js）

请先安装依赖：npm i -D ws

/**
 * 跟踪远程日志文件
 * @param {*} ps
 */
const _tailRemoteFile = async ps=>{
    let url = remoteUrl("/ws/file-tail")
    let index = url.indexOf("://")

    let headers = {}
    headers.params = JSON.stringify(ps)

    const client = new WebSocket(`ws${url.substring(index)}`, { headers })
    client.on('open', ()=> console.debug(chalk.magenta(`与服务器连接成功 🤝`)))
    // client.on('close',()=> console.debug(chalk.magenta(`\n与服务器连接关闭 👋`)))
    client.on('error', e=> {
        console.debug(chalk.red(e))
    })
    client.on('message', /** @param {Buffer} buf */buf=>{
        let line = buf.toString()
        if(line.endsWith("\n") || line.endsWith("\r\n"))
            line = line.substring(0, line.length-2)
        console.debug(line)
    })
}