websocket实现方法日志实时查询过程_websocket

本次方法的核心概念是通过redis生成唯一key值（没有放出来），然后通过前端获取这个唯一的key带入到方法请求中，然后服务器通过这个key生成此次方法生成唯一的日志文件，websocket接口通过线程实时读取key文件返回内容。

1、通过logback生成日志工具类

package utils;

import ch.qos.logback.classic.logger;
import ch.qos.logback.classic.loggercontext;
import ch.qos.logback.classic.encoder.patternlayoutencoder;
import ch.qos.logback.core.rolling.rollingfileappender;
import ch.qos.logback.core.rolling.sizeandtimebasedrollingpolicy;
import ch.qos.logback.core.util.filesize;
import ch.qos.logback.core.util.optionhelper;
import cn.sunline.dds.common.framework.constants.socketlog;
import org.slf4j.loggerfactory;

import java.util.hashmap;
import java.util.map;

/**
 * 日志构建器
 * @author mr.ye
 * @description: 该方法不会在控制台打印日志，只用于生成指定文件的日志
 */
public class loggerbuilder {  
  
    private static final map<string,logger> container = new hashmap<>();

    private static socketlog socketlog;

    public static logger getlogger(string key) {
        logger logger = container.get(key);
        if(logger != null) {
            return logger;
        }
        synchronized (loggerbuilder.class) {
            logger = container.get(key);
            if(logger != null) {
                return logger;
            }
            logger = build(key);
            container.put(key,logger);
        }
        return logger;
    }
    //删除container中的log
    public static void close(string key) {
        if (container.containskey(key)) {
            container.remove(key);
        }
    }
  
    private static logger build(string key) {
        loggercontext context = (loggercontext) loggerfactory.getiloggerfactory();  
        logger logger = context.getlogger("filelogger");
        logger.setadditive(false);  
        rollingfileappender appender = new rollingfileappender();  
        appender.setcontext(context);  
        appender.setname("filelogger");
        appender.setfile(optionhelper.substvars("/logs/"+ key + ".log",context));
        appender.setappend(true);  
        appender.setprudent(false);
        //重命名日志文件
        sizeandtimebasedrollingpolicy policy = new sizeandtimebasedrollingpolicy();
        string fp = optionhelper.substvars("/logs/" + key + ".log.%d{yyyy-mm-dd}.%i",context);
        policy.setmaxfilesize(filesize.valueof("128mb"));
        policy.setfilenamepattern(fp);
        policy.setmaxhistory(7);
        policy.settotalsizecap(filesize.valueof("32gb"));
        policy.setparent(appender);
        policy.setcontext(context);
        policy.start();
  
        patternlayoutencoder encoder = new patternlayoutencoder();  
        encoder.setcontext(context);  
        encoder.setpattern("%d{yyyy-mm-dd hh:mm:ss.sss} [%thread] %-5level %logger{20} - [%method,%line] - %msg%n");
        encoder.start();  
  
        appender.setrollingpolicy(policy);  
        appender.setencoder(encoder);  
        appender.start();  
        logger.addappender(appender);  
        return logger;  
    }
}

2、构建websocket方法

package socket;

import org.slf4j.logger;
import org.slf4j.loggerfactory;
import org.springframework.stereotype.component;

import javax.websocket.onclose;
import javax.websocket.onerror;
import javax.websocket.onopen;
import javax.websocket.session;
import javax.websocket.server.pathparam;
import javax.websocket.server.serverendpoint;
import java.io.ioexception;
import java.util.map;
import java.util.concurrent.concurrenthashmap;
import java.util.concurrent.atomic.atomicinteger;

/**
 * @data: 2022/04/22
 * @author mr.ye
 * @deprecated 由于读写文件占用资源较大，所以暂时不使用
 */
@serverendpoint("/log/{key}")
@component
public class logwebsocket {
    private final logger logger = loggerfactory.getlogger(logwebsocket.class);

    /**
     * 记录当前总共有多少个连接
     */
    private static final atomicinteger online_count = new atomicinteger(0);
    /**
     * 项目路径
     */
    private static final string property = system.getproperty("user.dir");

    /**
     * 存放当前正在做灵活开发的session对象
     */
    private static final map<string, session> clients = new concurrenthashmap<>();

    /**
     * 新的websocket请求开启
     * @describe 当前方法各系统通用，通过file获取日志文件，linux下可以使用tail -f持续获取日志文件更方便
     */
    @onopen
    public void onopen(session session,@pathparam("key")string key) {
        try {
            system.out.println("------------------key------------------:"+key);
            // 建立连接梳理 加 1
            online_count.incrementandget();
            // 将当前创建的session 存储起来
            clients.put(session.getid(), session);
            logger.info("有新窗口打开连接加入：{}，当前正在查询总数为：{}", session.getid(), online_count.get());
            string url = property+ "/logs/"+key+".log";
            logfiletailer tailer = new logfiletailer(url);
            tailer.addlistener(log -> {
                try {
//                    session.getbasicremote().sendtext(log + "<br />");
                    session.getbasicremote().sendtext(log);
                } catch (ioexception e) {
                    e.printstacktrace();
                }
            });
            tailer.start();
        } catch (exception e) {
            e.printstacktrace();
        }
    }
 
    /**
     * websocket请求关闭
     */
    @onclose
    public void onclose(session session) {
        // 建立连接梳理 减 1
        online_count.decrementandget();
        // 删除缓存起来的session
        clients.remove(session.getid());
        logger.info("有一窗口连接关闭：{}，当前正在开发总数为：{}", session.getid(), online_count.get());
    }

    @onerror
    public void onerror(session session, throwable error) {
        error.printstacktrace();
        // 建立连接梳理 减 1
        online_count.decrementandget();
        // 删除缓存起来的session
        clients.remove(session.getid());
        logger.error("发生错误，删除当前session：{}，剩余正在总数为：{}", session.getid(), online_count.get());
    }
}

3、日志监听获取

package socket;

import java.io.file;
import java.io.ioexception;
import java.io.randomaccessfile;
import java.util.function.consumer;

/**
 * @data: 2022/04/24
 * @author mr.ye
 * @deprecated 由于读写文件占用资源较大，所以暂时不使用
 */
public class logfiletailer extends thread {

	private file logfile;

	private consumer<string> callback;

	/**
	 * 监视开关，true = 打开监视
	 */
	private boolean tailing = true;

	/**
	 * 
	 * @param file 要监视的文本文件
	 */
	public logfiletailer(string file) {
		logfile = new file(file);

	}

	/**
	 *tailing 开关
	 * @param tailing
	 */
	public void tailing(boolean tailing) {
		this.tailing = tailing;

	}

	/**
	 * 设置回调事件
	 * 
	 * @param callback 回调事件
	 */
	public void addlistener(consumer<string> callback) {
		this.callback = callback;
	}

	@override
	public void run() {
		/*上一次读取文件位置*/
		long filepointer =0;
		try {
			randomaccessfile file  = new randomaccessfile(logfile, "r");
			while (tailing) {
				long filelength = logfile.length();
				if (filelength < filepointer) {
					file = new randomaccessfile(logfile, "r");
					filepointer = 0;
				}
				if (filelength > filepointer) {
					file.seek(filepointer);
					string line = file.readline();

					while (line != null) {
						line = new string(line.getbytes("iso-8859-1"), "utf-8");
						if (callback != null){
							callback.accept(line);
						}
						line = file.readline();
					}

					filepointer = file.getfilepointer();
				}

//				sleep(sampleinterval);
			}

			file.close();
		} catch (ioexception e) {
			e.printstacktrace();
		}
	}
}

4、问题补充

springboot中需要注入bean，通过注入一个serverendpointexporter,该bean会自动注册使用@serverendpoint注解申明的websocket endpoint。

要注意：如果使用独立的servlet容器，而不是直接使用springboot的内置容器，

就不要注入serverendpointexporter，因为它将由容器自己提供和管理，所以在注入

serverendpointexporter时需要增加一些配置

package conf;

import org.springframework.boot.web.servlet.servletcontextinitializer;
import org.springframework.context.annotation.bean;
import org.springframework.context.annotation.configuration;
import org.springframework.web.socket.server.standard.serverendpointexporter;

import javax.servlet.servletcontext;
import javax.servlet.servletexception;

/**
 * 注入一个serverendpointexporter,
 * 该bean会自动注册使用@serverendpoint注解申明的websocket endpoint
 * @author mr.ye
 *
 */
@configuration
public class websocketconfig implements servletcontextinitializer {

    /**
     * 要注意，如果使用独立的servlet容器，而不是直接使用springboot的内置容器，
     * 就不要注入serverendpointexporter，因为它将由容器自己提供和管理
     * 所以增加以下配置：
     * @return
     */
	@bean
    public serverendpointexporter serverendpointexporter() {
        return new serverendpointexporter();
    }

    @override
    public void onstartup(servletcontext servletcontext) throws servletexception {
        servletcontext.setinitparameter("org.apache.tomcat.websocket.textbuffersize", string.valueof(10 * 1024 * 1024));
        servletcontext.setinitparameter("org.apache.tomcat.websocket.binarybuffersize", string.valueof(10 * 1024 * 1024));
    }
}