观察者模式小练习：多日志监控 by 嗡汤圆

女爷i 2022-09-25 05:11 225阅读 0赞

想写个小工具包用来在web项目中提供实时查看各系统日志输出的功能。以下仅仅是小练手。  
具体特性如下：  
1. 观察者模式实现  
2. 任意多文件实时捕获输出  
3. 多用户查看同一文件时，维持单一读取线程，保证IO效率  
4. 当无用户查看某文件时，自动停止对该文件的监控

# 说明 #

若想简单实现查看某文件的变化，只要写个循环一直读取就好了，但是这样会造成一个问题：用户少的情况下，少数线程循环读取文件速度变化不明显，若用户数量增多则会出现性能问题。即使假设有10000个用户只读取1个文件，这10000个线程均会循环读取文件，造成磁盘反复在不同位置跳动，严重影响IO性能，因此需要对磁盘性能资源进行统一管理。

# 设计 #

## 观察者模式的基本实现 ##

借助JDK自己的Observable和Observer即可。

### 监视文件的Observable ###

该类既有Observable的特性，且需要具备循环运行的能力，因此也要实现Runnable接口。  
代码如下：

public class TailSubject extends Observable implements Runnable { 
        private static final Logger logger = LoggerFactory.getLogger(TailSubject.class);
        private File logFile;/*日志文件位置*/
        private long sampleInterval = 200;/*默认读取延迟*/
        private boolean startFromTop = false;/*是否从头开始*/
        private boolean tailing = false;/*是否循环读取*/
        public TailSubject(String filePath) {
            this.logFile = new File(filePath);
        }
        public TailSubject(File file) {
            this.logFile = file;
        }
        public void setInterval(long interval){
            this.sampleInterval = interval;
        }
        public void setStartFromTop(boolean flag){
            this.startFromTop = flag;
        }
        public void startTailing() {
            logger.info("start tailing");
            this.tailing = true;
        }
        public void kill() {
            logger.info("stop tailing");
            this.tailing = false;
        }
        public void run() {
            logger.info("thread started");
            long filePointer = 0;
            if (this.startFromTop) {
                logger.info("start from top");
                filePointer = 0;
            } else {
                logger.info("start from " + this.logFile.length());
                filePointer = this.logFile.length();
            }
            try {
                RandomAccessFile file = new RandomAccessFile(this.logFile, "r");/* readOnlyFile */
                while (this.tailing) {
                    long fileLength = this.logFile.length();
                    if (fileLength < filePointer) {
                        logger.info("log file length reduced.");
                    }
                    if (fileLength > filePointer) {
                        file.seek(filePointer);
                        String line = file.readLine();
                        while (line != null) {
                            /*在这里通知观察者*/
                            setChanged();/*setChanged此步骤必须*/
                            notifyObservers(line);
                            line = file.readLine();
                        }
                        filePointer = file.getFilePointer();
                    }
                    Thread.currentThread();
                    Thread.sleep(this.sampleInterval);
                }
            } catch (IOException e) {
                logger.error(e.toString());
            } catch (InterruptedException e) {
                logger.error(e.toString());
            }
        }
    }

### Observer的基本实现 ###

直接集成Observer类即可，需要重写update方法。

// 以更新时记录日志的观察者和屏幕输出的观察者为例，如下：
    public class LoggingObserver implements Observer { 
        private static final Logger logger = LoggerFactory.getLogger(LoggingObserver.class);
        public void update(Observable o, Object arg) {
            logger.info("LoggingObserver updating:" + arg);
        }
    }
    public class SysoutObserver implements Observer { 
        public void update(Observable o, Object arg) {
            System.out.println("SysoutObserver updating:" + arg);
        }
    }

## Observable的统一管理 ##

单例模式建立一个Observable资源的管理类，确保一个文件仅有一个Observable在监视它，同时在没有观察者订阅时自动结束监视线程，以节省磁盘资源。

public class TailPool { 
    
        private static final Logger logger = LoggerFactory.getLogger(TailPool.class);
        private static HashMap<String, TailSubject> subjects;
        public static volatile TailPool instance;
    
        private TailPool() {
            subjects = new HashMap<String, TailSubject>();
        }
    
        public static TailPool getInstance(){
            if(instance == null){
                synchronized (TailPool.class) {
                    if(instance == null){
                        instance = new TailPool();
                    }
                }
            }
            return instance;
        }
    
        /*注册Observer*/
        public synchronized TailSubject subscribe(String filePath){
            if(subjects.containsKey(filePath)){
                logger.info("subject already exist.");
                return subjects.get(filePath);
            }
            logger.info("creating new subject:"+filePath);
            TailSubject subject = new TailSubject(filePath);
            subject.startTailing();
            subjects.put(filePath, subject);
            new Thread(subject).start();
            return subject;
        }
        /*取消Observer观察*/
        public synchronized void unSubscribe(String filePath, Observer obs){
            TailSubject subject = subjects.get(filePath);
            if(subject == null){
                logger.info("subject doesn't exist");
            } else {
                subject.deleteObserver(obs);
                int observerCount = subject.countObservers();
                logger.info("subject has "+observerCount+" observers left.");
                if(observerCount == 0){
                    logger.info("no observer, stopping, remove subject from subject cache.");
                    subject.kill();
                    subjects.remove(filePath);
                }
            }
        }
    }

## 使用 ##

String file = "C://log1";
        LoggingObserver ob1 = new LoggingObserver();
        TailPool.getInstance().subscribe(file).addObserver(ob1);/*订阅*/
        TailPool.getInstance().unSubscribe(file, ob1);/*取消订阅*/

# 实例演示 #

这里演示订阅两个文件的监视，其中文件1有两个观察者（LoggingObserver，SysoutObserver），文件2只有一个LoggingObserver。在监视结束后陆续取消以上观察者的监听，并观察程序是否自动结束循环并退出。

## 启动、订阅 ##

String file = "C://log1";
    String file2 = "C://log2";
    LoggingObserver ob1 = new LoggingObserver();
    SysoutObserver ob2 = new SysoutObserver();
    LoggingObserver ob3 = new LoggingObserver();
    TailPool.getInstance().subscribe(file).addObserver(ob1);
    TailPool.getInstance().subscribe(file).addObserver(ob2);
    TailPool.getInstance().subscribe(file2).addObserver(ob3);

效果如下：  
![这里写图片描述][20160813010136152]  
可以看到第一个Observer注册的时候新建了一个文件监听线程，第二个Observer由于观察了同一个文件，因此直接在已有的线程注册一个自己，而不是新建线程“subject already exist”。  
第三个Observer观察不同的文件，因此有新建了一个线程。

## 日志文件输出 ##

通过记事本修改日志文件模拟文件的变化。效果如下  
![这里写图片描述][20160813010438453]  
可以看到修改第一个文件时，两个观察者都得到了通知，并分别以Log和system.out.println的形式进行了更新。修改第二个文件时，第三个观察者输出了更新。

## 停止观察并退出 ##

依次调用unSubscribe，观察程序是否退出。

System.in.read();
    logger.info("unsubscribing...");
    TailPool.getInstance().unSubscribe(file, ob1);
    TailPool.getInstance().unSubscribe(file, ob2);
    TailPool.getInstance().unSubscribe(file2, ob3);

手动敲一个回车即可开始停止的过程。分别依次取消各个观察者的监听。  
![这里写图片描述][20160813010754638]  
可以看到第一个文件在Observer1取消时由于还有Observer2，因此不会退出。在Observer2也取消后，线程自行退出循环。同理Observer3取消时，线程2也自行退出了循环。此时已没有其余线程在运行，程序结束退出。

[20160813010136152]: /images/20220718/9c13d91bef5b460cbb2cfe104db603c2.png
[20160813010438453]: /images/20220718/803f8fa993ad4807bba64b1e8ee5dec8.png
[20160813010754638]: /images/20220718/ece0aab78a4b45d7bb3bfc76ac651acc.png