美文网首页
项目实战—观察者模式(自定义高性能的订阅-发布模型)

项目实战—观察者模式(自定义高性能的订阅-发布模型)

作者: 小胖学编程 | 来源:发表于2021-01-28 21:40 被阅读0次

    观察者模式是对象的行为模式,又叫发布-订阅(Publish/Subscribe)模式、模型-视图(Model/View)模式、源-监听器(Source/Listener)模式或从属者(Dependents)模式。

    观察者模式定义了一种一对多的依赖关系,让多个观察者对象同时监听某一个主题对象。这个主题对象在状态上发生变化时,会通知所有观察者对象,使它们能够自动更新自己。

    1. JDK提供的观察者模式

    在JAVA语言的java.util库里面,提供了一个Observable类以及一个Observer接口,构成JAVA语言对观察者模式的支持。

    事件对象:

    @Data
    public class NameEvent {
    
        private String name;
    
        //信息
        private String message;
    }
    

    发布者(被观察者):

    import java.util.Observable;
    import java.util.Observer;
    
    public class BusService {
    
        //发布者对象
        private NameWatched observable = new NameWatched();
        
        public void addObservers(Observer... observers) {
            //支持动态新增观察者
            for (Observer observer : observers) {
                observable.addObserver(observer);
            }
        }
    
        /**
         * 业务方法
         */
        public void bus(String name) {
            //当名字为tom时,触发事件
            if ("tom".equals(name)) {
                NameEvent event = new NameEvent();
                event.setName("tom");
                event.setMessage("触发事件啦!");
                observable.sendNameEvent(event);
            }
    
        }
    
        /**
         * 发布者对象(观察者)
         */
        public static class NameWatched extends Observable{
    
            public void sendNameEvent(NameEvent nameEvent){
                //发送变化了
                setChanged();
                //推送消息
                notifyObservers(nameEvent);
            }
    
        }
    
    }
    

    订阅者(观察者):

    @Slf4j
    public class NameWatcher implements Observer {
        @Override
        public void update(Observable o, Object arg) {
            //获取到业务参数
          log.info("监听者-打印数据{}", JSON.toJSONString(arg));
        }
    }
    

    测试代码:

    public class Testlucene {
        public static void main(String[] args) {
    
            //线程1,执行业务逻辑
            new Thread(()->{
                BusService busService=new BusService();
                busService.addObservers(new NameWatcher());
                busService.bus("tom");
            }).start();
    
    
            try {
                Thread.sleep(10000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }
    

    结果:

    21:15:29.258 [Thread-0] INFO com.tellme.test.NameWatcher - 监听者-打印数据{"message":"触发事件啦!","name":"tom"}
    

    缺点:

    1. 传递参数的方式在notifyObservers加了sync(this)关键字,会严重影响性能,但作用仅仅就是同步获取增删的订阅者。
    2. 传递this对象,不能声明全局的Observable对象。
    3. 多个订阅者和发布者时同一个线程。会影响发布者性能!

    2. Google Guava的EventBus提供的观察者模式

    使用方式:
    JAVA进阶篇(10)—Guava实现的EventBus(观察者模式)

    缺点:

    1. EventBus需要注意:发布者和订阅者使用同一个线程,可能会影响发布者的性能。但可以保证单线程中事件的发布顺序和调度顺序保持一致。
    2. AsyncEventBus需要注意的是:发布者和订阅者可以使用不同的线程处理;发布事件时维护了一个LinkedQueue,若订阅者消费速度慢,可能会造成内存溢出;采用全局队列维护事件顺序性,但不能完全保证调度和发布的顺序;性能不如直接分发好;
    3. guava的EventBus虽然通过注解的方式更加灵活,但是没有接口的语法层面的依赖关系,代码维护性、可读性不是特别好。

    3. 自定义实现

    public class SealObservable {
    
        /**
         * 线程安全,且适合读多写少的场景
         */
        private List<SealObserver> sealObservers = new CopyOnWriteArrayList<>();
    
        private Executor executor;
    
        public SealObservable() {
            executor = DirectExecutor.INSTANCE;
        }
    
        /**
         * @param sync true:使用同一个线程处理消息。false:开启异步线程处理消息
         */
        public SealObservable(boolean sync) {
            if (sync) {
                executor = DirectExecutor.INSTANCE;
            } else {
                executor = new ThreadPoolExecutor(4,
                        8, 60, TimeUnit.SECONDS,
                        new ArrayBlockingQueue<>(10), new ThreadPoolExecutor.DiscardPolicy());
            }
        }
    
        public SealObservable(Executor executor) {
            this.executor = executor;
        }
    
        public void addObserver(SealObserver o) {
            if (o == null)
                throw new NullPointerException();
            if (!sealObservers.contains(o)) {
                sealObservers.add(o);
            }
        }
    
        public void deleteObserver(SealObserver o) {
            sealObservers.remove(o);
        }
    
        /**
         * 通知消息
         */
        public void notifyObservers(Object arg) {
            //没有监听者,快速失败
            if (sealObservers.size() == 0) {
                return;
            }
            //开启线程配置
            for (SealObserver sealObserver : sealObservers) {
                executor.execute(() -> {
                    sealObserver.notice(arg);
                });
            }
        }
    
    }
    

    直接线程池:

    /**
     * 直接线程池。
     */
    public enum DirectExecutor implements Executor {
        INSTANCE;
    
        @Override
        public void execute(Runnable command) {
            command.run();
        }
    
        @Override
        public String toString() {
            return "SealExecutors.directExecutor()";
        }
    }
    

    观察者对象:

    public interface SealObserver {
    
        /**
         * 触发事件调用的方法
         *
         * @param arg 事件对象
         */
        void notice(Object arg);
    }
    

    相关文章

      网友评论

          本文标题:项目实战—观察者模式(自定义高性能的订阅-发布模型)

          本文链接:https://www.haomeiwen.com/subject/qzextltx.html