SpringBoot 整合websocket 实现日志实时查看

大家好，我是一安

某一天客户提出想直接通过页面实时地看到各个程序运行的过程，也就是将程序处理过程的日志实时展现给前端，我首先想到的方案就是通过websocket的方式

其实主要就是分为以下几个步骤

用户点击查看日志按钮，与后端进行通道连接，监听日志文件变化
将变化的内容通过websocket 发送到前端
用户关闭窗口，关闭监听

介绍

WebSocket是一种在单个TCP连接上进行全双工通信的协议。WebSocket使得客户端和服务器之间的数据交换变得更加简单，允许服务端主动向客户端推送数据。在WebSocket API中，浏览器和服务器只需要完成一次握手，两者之间就可以直接创建持久性的连接，并进行双向数据传输。

正文

环境准备

注意：

1.在websocket 中使用antowired 无效，可以自定义一个SpringContextUtils获取，或者使用构造方法注入

2.要保证SpringContextUtils在当前websocket之前加载，可以使用@DependsOn(value = “springContextUtils”)进行修饰

3.spring 给每个session会话都会创建一个websocket实例，如果需要共享变量，可以使用static修饰

1. 引入依赖

    <!-- SSH-2协议-->
    <dependency>
        <groupId>ch.ethz.ganymed</groupId>
        <artifactId>ganymed-ssh2</artifactId>
        <version>build210</version>
    </dependency>
    <!--websocket日志预览-->
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-websocket</artifactId>
    </dependency>

2. 配置文件

/**
 * 注入一个ServerEndpointExporter,该Bean会自动注册使用@ServerEndpoint注解申明的websocket endpoint
 */
@Configuration
public class WebSocketConfig {
    @Bean
    public ServerEndpointExporter serverEndpointExporter() {
        return new ServerEndpointExporter();
    }
}

3. 核心处理过程，用来进行服务端与客户端之间交互

实际过程可以根据参数读取数据库获取设备信息，小编这里为了简便测试直接写的默认

@Component
@CrossOrigin
@ServerEndpoint(value = "/log")
public class LogWebSocketHandle {

    private static Logger logger = LoggerFactory.getLogger(LogWebSocketHandle.class);

    //注入的时候,给类的 service 注入,否则无法注入
    private static JdbcTemplate MySQLJdbc;
    @Autowired
    public void setChatService(@Qualifier("mysqlJdbcTemplate") JdbcTemplate jdbcTemplate) {
        LogWebSocketHandle.mysqlJdbc = jdbcTemplate;
    }

    private Session ssh = null;
    private Connection conn = null;
    private final int port = 22;

    /**
     * 新的WebSocket请求开启
     */
    @OnOpen
    public void onOpen(Javax.websocket.Session session) {
        try {
//            //应用类型
//            List<String> listMap = session.getRequestParameterMap().get("type");
//            String type = listMap.get(0);
//            //应用名称
//            List<String> listMap2 = session.getRequestParameterMap().get("name");
//            String name = listMap2.get(0);
//
//            String sql = "select d.*,da.logfilepath from D_APP d,D_APP_"+type.toUpperCase()+
//                    " da where d.id=da.appid and d.name=?";
//
//            List<Map<String,String>>list= mysqlJdbc.query(sql, new Object[]{name}, new RowMapper<Map<String, String>>() {
//                @Override
//                public Map<String, String> mapRow(ResultSet rs, int i) throws SQLException {
//                    Map<String,String> map = new HashMap<String,String>();
//                    String ip = rs.getString("ip");
//                    String username = rs.getString("username");
//                    String password = rs.getString("password");
//                    String logpath = rs.getString("logfilepath");
//                    map.put("ip",ip);
//                    map.put("username",username);
//                    map.put("password",password);
//                    map.put("logpath",logpath);
//                    return map;
//                }
//            });
            Map<String,String> map = new HashMap<>();
            map.put("ip","xxx.xxx.xxx.xxx");
            map.put("username","xxxx");
            map.put("password","xxxxxx!");
            map.put("logpath","/usr/local/xxxx/5gvpdn/restapi/xxxx/logs/common-2022-09-02-0.txt");
            List<Map<String,String>>list = new ArrayList<>();
            list.add(map);
            if(list.size()>0){
                Map<String,String> map2 = list.get(0);
                /** 服务器参数 */
                String hostname = map2.get("ip");
                String username = map2.get("username");
                String password = map2.get("password");
                String logpath = map2.get("logpath");
                conn = new Connection(hostname, port);
                //连接到主机
                conn.connect();
                //使用用户名和密码校验
                boolean isconn = conn.authenticateWithPassword(username, password);
                if (!isconn) {
                    session.getBasicRemote().sendText("用户名称或者是密码不正确" + "<br>");
                } else {
                    session.getBasicRemote().sendText(hostname + "：连接成功!" + "<br>");
                    ssh = conn.openSession();
                    ssh.execCommand("tail -f " + logpath);
                    InputStream inputStream = new StreamGobbler(ssh.getStdout());
                    TailLogThread thread = new TailLogThread(inputStream, session);
                    thread.start();
                }
            }
        } catch (Exception e) {
            logger.error(e.getMessage(), e);
            try {
                session.getBasicRemote().sendText(e.getMessage() + "<br>");
            } catch (Exception e1) {
                logger.error(e1.getMessage(), e1);
            }
        }
    }


    @OnClose
    public void onClose() {
        try {
            if (ssh != null) {
                ssh.close();
            }
            if (conn != null) {
                conn.close();
            }
        } catch (Exception e) {
            logger.error(e.getMessage(), e);
        }
    }

    @OnError
    public void onError(Throwable thr) {
        logger.error(thr.getMessage(), thr);
    }
}

4. 文件监听使用异步，否则会导致占用主线程，导致无法断开连接

public class TailLogThread extends Thread {

    private static Logger logger = LoggerFactory.getLogger(TailLogThread.class);

    private BufferedReader reader;
    private Session session;

    public TailLogThread(InputStream in, Session session) {
        this.reader = new BufferedReader(new InputStreamReader(in));
        this.session = session;

    }

    @Override
    public void run() {
        String line;
        try {
            while ((line = reader.readLine()) != null) {
                // 将实时日志通过WebSocket发送给客户端，给每一行添加一个HTML换行
                session.getBasicRemote().sendText(line + "<br>");
            }
        } catch (EOFException e1) {
            try {
                session.getBasicRemote().sendText("客户端已经关闭!" + "<br>");
            } catch (Exception e) {
                logger.error("服务流关闭提示消息发送报错：" + e.getMessage(), e);
            }
            logger.info("出现了EOFExcption:：服务流已经关闭!");
        } catch (Exception e) {
            logger.error(e.getMessage(), e);
        }
    }
}