`

观察者模式 and Eventbus 转帖

阅读更多

转帖地址:https://www.cnblogs.com/zemliu/archive/2013/09/11/3313782.html

观察者模式

结构图

 

代码实现

复制代码
public abstract class Subject {

    private List<Observer> observerList = new ArrayList<Observer>();

    /**
     * 注册观察者
     * @param observer
     */
    public void register(Observer observer) {
        observerList.add(observer);
    }

    /**
     * 注销观察者
     *
     * @param observer
     */
    public void unregister(Observer observer) {
        observerList.remove(observer);
    }

    /**
     * 通知观察者更新
     */
    public void post() {
        for (Observer observer : observerList) {
            observer.update();
        }
    }

    /**
     * 获取被通知事件
     *
     * @return
     */
    public abstract Object getEvent();
}


public class ConcreteSubject1 extends Subject {

    /** 个性化的定制内容 */
    private String subjectState;

    public ConcreteSubject1(String subjectState) {
        this.subjectState = subjectState;
    }

    @Override
    public Object getEvent() {
        System.out.println("Custom ConcreteSubject1");
        return subjectState;
    }
}


public class ConcreteSubject2 extends Subject {

    private int subjectState;

    public ConcreteSubject2(int subjectState) {
        this.subjectState = subjectState;
    }

    @Override
    public Object getEvent() {
        System.out.println("Custom ConcreteSubject2");
        return subjectState;
    }
}


public abstract class Observer {

    /** 用于观察者获取被通知的事件 */
    protected Subject subject;

    /**
     * 用于给Subject通知时调用的更新方法
     */
    public abstract void update();
}


public class ConcreteObserver1 extends Observer {

    public ConcreteObserver1(Subject subject) {
        this.subject = subject;
    }

    @Override
    public void update() {
        System.out.println("Subject " + subject.getEvent() + " ConcreteObserver1");
    }
}


public class ConcreteObserver1 extends Observer {

    public ConcreteObserver1(Subject subject) {
        this.subject = subject;
    }

    @Override
    public void update() {
        System.out.println("Subject " + subject.getEvent() + " ConcreteObserver1");
    }
}


public class ConcreteObserver1 extends Observer {

    public ConcreteObserver1(Subject subject) {
        this.subject = subject;
    }

    @Override
    public void update() {
        System.out.println("Subject " + subject.getEvent() + " ConcreteObserver1");
    }
}
复制代码

 

输出:

Custom ConcreteSubject1
Subject Sub1 ConcreteObserver1
Custom ConcreteSubject1
Subject Sub1 ConcreteObserver2

 

 

EventBus简单示例

EventBus是Guava提供的消息发布-订阅类库,它的工作机制类似于观察者模式,通过通知者去注册观察者,最后由通知者想观察者发布消息,示例代码

复制代码
public class MsgCenter {

    /** EventBus的定位跟接近于消息中心,而他的post()方法跟接近于一个自定义的Subject */
    public static EventBus eventBus = new EventBus();
}


public class Observer1 {

    /**
     * 只有通过@Subscribe注解的方法才会被注册进EventBus
     * 而且方法有且只能有1个参数
     * @param msg
     */
    @Subscribe
    public void ob1Mthod1(String msg) {
        System.out.println(msg + " test1!");
    }

    @Subscribe
    public void ob1Method2(String msg) {
        System.out.println(msg + " test2!");
    }
}


public class Observer2 {
    @Subscribe
    public void ob2Method1(String msg) {
        System.out.println(msg + " test3!");
    }

    // 错误的基本型参数
    // @Subscribe
    // public void ob2Method2(int msg) {
    // System.out.println(msg + " test4!");
    // }
    /**
     * post() 不支持自动装箱功能,只能使用Integer,不能使用int,否则handlersByType的Class会是int而不是Intege
     * 而传入的int msg参数在post(int msg)的时候会被包装成Integer,导致无法匹配到
     */
    @Subscribe
    public void ob2Method2(Integer msg) {
        System.out.println(msg + " test4!");
    }
}



public class Test {

    public static void main(String[] args) throws InterruptedException {
        EventBus eventBus = new EventBus();
        Observer1 observer1 = new Observer1();
        Observer2 observer2 = new Observer2();

        eventBus.register(observer1);
        eventBus.register(observer2);

        // 只有注册的参数类型为String的方法会被调用
        eventBus.post("post string method");

        // 注销observer2
        eventBus.unregister(observer2);
        eventBus.post("post string method after unregister");
    }
}
复制代码

 

输出

post string method test2!
post string method test1!
post string method test3!
post string method after unregister test2!
post string method after unregister test1!

 

实际上EventBus要表达的意图很简单,就是将post(Object arg)这里的arg当做参数传入到已注册的方法(被@Subscribe)的方法里,并调用该方法,所以当post(String)的时候,调用的参数类型为String的注册方法,当post(int)的时候,调用则是参数类型为Integer的注册方法

 

EventBus的实现方式

eventbus的实现方式实际上类似于上例写的简单的观察者模式,不同点在于它实现了泛化的注册方法以及泛化的方法调用,另外还考虑到了多线程的问题,对多线程使用时做了一些优化

register(Object listener)

register()方法注册一个任意类型的实例并将其使用了@Subscribe的方法注册到一个Map中,这个Map使用方法的参数类型Class为Key,值为一个Set,这个Set包含了所有参数类型为Key的EventHandler,Eventhandler是EventBus定义的一个数据结构,由listener(方法拥有者instance,例如上例中的observer1)和这个listener的@Subscribe方法构成

]这个Map的结构示意图如下

 

其实现代码如下

复制代码
  /**
   * Registers all handler methods on {@code object} to receive events.
   * Handler methods are selected and classified using this EventBus's
   * {@link HandlerFindingStrategy}; the default strategy is the
   * {@link AnnotatedHandlerFinder}.
   *
   * @param object  object whose handler methods should be registered.
   */
  public void register(Object object) {
    handlersByType.putAll(finder.findAllHandlers(object));
  }



  /**
   * {@inheritDoc}
   *
   * This implementation finds all methods marked with a {@link Subscribe} annotation.
   */
  @Override
  public Multimap<Class<?>, EventHandler> findAllHandlers(Object listener) {
    Multimap<Class<?>, EventHandler> methodsInListener = HashMultimap.create();
    Class<?> clazz = listener.getClass();
    Set<? extends Class<?>> supers = TypeToken.of(clazz).getTypes().rawTypes();

    for (Method method : clazz.getMethods()) {
      /*
       * Iterate over each distinct method of {@code clazz}, checking if it is annotated with
       * @Subscribe by any of the superclasses or superinterfaces that declare it.
       */
      for (Class<?> c : supers) {
        try {
          Method m = c.getMethod(method.getName(), method.getParameterTypes());
          if (m.isAnnotationPresent(Subscribe.class)) {
            Class<?>[] parameterTypes = method.getParameterTypes();
            if (parameterTypes.length != 1) {
              throw new IllegalArgumentException("Method " + method
                  + " has @Subscribe annotation, but requires " + parameterTypes.length
                  + " arguments.  Event handler methods must require a single argument.");
            }
            Class<?> eventType = parameterTypes[0];
            EventHandler handler = makeHandler(listener, method);

            methodsInListener.put(eventType, handler);
            break;
          }
        } catch (NoSuchMethodException ignored) {
          // Move on.
        }
      }
    }
    return methodsInListener;
  }
复制代码

 

post(Object event)

post(Object event)方法用于向已注册的方法传递一个参数,并以此调用参数类型为event.class的所有方法,调用的时候会使用一个ThreadLocal的Queue来进行任务分发,这样的结果就是在多线程情况下,线程间共享注册方法的Map(上面提到那个),当时在发送消息时线程会保有自己独立的一个Post任务的Queue,保证了线程执行post()方法时候的独立性而不会相互影响,下面是多线程执行post()时的示意图

真正在执行post(Object event)的时候,会将msg与所有event.class对应的set里的所有method组合成一个EventBus.EventWithHandler对象并加入到ThreadLocal的Queue中,最后再将Queue出队依次执行这些方法,最后清空ThreadLocal的Queue,EventBus的实现代码如下

复制代码
    /**
     * Posts an event to all registered handlers.  This method will return
     * successfully after the event has been posted to all handlers, and
     * regardless of any exceptions thrown by handlers.
     *
     * <p>If no handlers have been subscribed for {@code event}'s class, and
     * {@code event} is not already a {@link com.google.common.eventbus.DeadEvent}, it will be wrapped in a
     * DeadEvent and reposted.
     *
     * @param event  event to post.
     */
    @SuppressWarnings("deprecation") // only deprecated for external subclasses
    public void post(Object event) {
        Set<Class<?>> dispatchTypes = flattenHierarchy(event.getClass());

        boolean dispatched = false;
        // 将event和所有event.class对应的方法组合成EventWithHandler并入队
        for (Class<?> eventType : dispatchTypes) {
            Set<EventHandler> wrappers = getHandlersForEventType(eventType);

            if (wrappers != null && !wrappers.isEmpty()) {
                dispatched = true;
                for (EventHandler wrapper : wrappers) {
                    enqueueEvent(event, wrapper);
                }
            }
        }

        if (!dispatched && !(event instanceof DeadEvent)) {
            post(new DeadEvent(this, event));
        }

        dispatchQueuedEvents();
    }

    /**
     * Queue the {@code event} for dispatch during
     * {@link #dispatchQueuedEvents()}. Events are queued in-order of occurrence
     * so they can be dispatched in the same order.
     */
    void enqueueEvent(Object event, EventHandler handler) {
        eventsToDispatch.get().offer(new EventWithHandler(event, handler));
    }

    /**
     * Drain the queue of events to be dispatched. As the queue is being drained,
     * new events may be posted to the end of the queue.
     *
     * @deprecated This method should not be overridden outside of the eventbus package. It is
     *     scheduled for removal in Guava 14.0.
     */
    @Deprecated
    protected void dispatchQueuedEvents() {
        // don't dispatch if we're already dispatching, that would allow reentrancy
        // and out-of-order events. Instead, leave the events to be dispatched
        // after the in-progress dispatch is complete.
        if (isDispatching.get()) {
            return;
        }

        isDispatching.set(true);
        try {
            while (true) {
                EventWithHandler eventWithHandler = eventsToDispatch.get().poll();
                if (eventWithHandler == null) {
                    break;
                }

                dispatch(eventWithHandler.event, eventWithHandler.handler);
            }
        } finally {
            isDispatching.set(false);
        }
    }
复制代码

斜体加粗部分即为关键部分

 

EventBus多线程使用示例

复制代码
public class Test2 {

    public static void main(String[] args) throws InterruptedException {

        Thread t1 = new Thread() {

            @Override
            public void run() {
                System.out.println(Thread.currentThread());
                System.out.println("start");
                Observer1 observer1 = new Observer1();
                MsgCenter.eventBus.register(observer1);
                MsgCenter.eventBus.post("post string");
                System.out.println("end");
                System.out.println();
            }
        };

        Thread t2 = new Thread() {

            @Override
            public void run() {
                System.out.println(Thread.currentThread());
                System.out.println("start");
                Observer2 observer2 = new Observer2();
                MsgCenter.eventBus.register(observer2);
                MsgCenter.eventBus.post("post string2");
                System.out.println("end");
                System.out.println();
            }
        };

        ExecutorService exec = Executors.newFixedThreadPool(2);

        exec.execute(t1);

        // 为何忽略多线程是run()方法的线程安全问题,让两个任务分开执行
        TimeUnit.MILLISECONDS.sleep(500);

        exec.execute(t2);

        exec.shutdown();
    }
}
复制代码

输出

Thread[pool-1-thread-1,5,main]
start
post string test1!
post string test2!
end

Thread[pool-1-thread-2,5,main]
start
post string2 test1!
post string2 test2!
post string2 test3!
end

这里为了让执行结果更清晰,并没有让两个线程并发执行,但可以清楚看到他们是共享同一个注册方法的Map的,而由于post()分发消息时间的Queue是ThreadLocal的,这个Queue由每个线程独有

对于这里使用ThreadLocal的Queue的个人理解就是假如不使用ThreadLocal,共享同一个队列,就有可能由Thread1入队的EventWithHandler会由Thread2来执行,而由Thread2入队的EventWithHandler又有可能会由Thread1来执行,而造成执行秩序混乱.

而使用ThreadLocal的Queue,则这些EventWithHandler示例是由哪个线程入队就由哪个线程执行,同时不需要去考虑共享队列的入队和出队时的线程安全问题,也可以提升效率

分享到:
评论

相关推荐

    android观察者模式:EventBus实现GPS定位

    EventBus是一款轻量级的事件总线库,它是Android实现观察者模式的一个优秀工具,特别适合解决组件之间的通信问题。下面我们将深入探讨如何使用EventBus实现GPS定位功能。 首先,理解EventBus的基本概念。EventBus是...

    57丨观察者模式(下):如何实现一个异步非阻塞的EventBus框架?1

    在本节课程中,我们将深入探讨如何实现一个异步非阻塞的EventBus框架,以进一步理解观察者模式的应用。观察者模式是一种行为设计模式,它定义了对象间的一种一对多依赖关系,当一个对象的状态发生改变时,所有依赖于...

    跟我一起学 设计模式-观察者模式

    ASP.NET MVC和ASP.NET Core中的事件总线(EventBus)也是一种观察者模式的应用。 4. **数据库事件**:在SQL数据库中,可以使用触发器(Triggers)来实现类似观察者的效果,当数据表中的数据发生变化时,触发器会...

    android学习日记-6 观察者模式与订阅者模式

    在Android开发中,观察者模式(Observer Pattern)和订阅者模式(Subscriber Pattern)是两种重要的设计模式,它们在实现事件驱动编程中起着关键作用。本文将深入探讨这两种模式的概念、工作原理以及在Android开发中...

    观察者模式

    9. **扩展**:除了Java内置的`Observer`和`Observable`,开发者还可以使用第三方库如Guava的`EventBus`,或者Spring框架的`ApplicationEvent`和`ApplicationListener`,这些都提供了更高级的观察者模式实现。...

    Android设计模式之观察者模式

    **Android设计模式之观察者模式** 观察者模式(Observer Pattern)是软件设计模式中的行为模式之一,它在Android开发中有着广泛的应用。该模式的主要目的是定义对象之间的一种一对多依赖关系,当一个对象的状态发生...

    跟着 Guava、Spring 学习如何设计观察者模式.doc

    Guava的EventBus是观察者模式的一个实例,它简化了事件发布和订阅的过程。EventBus允许你定义事件的发布者(即被观察者)和事件的订阅者(即观察者)。发布者通过调用EventBus的post方法发布事件,而订阅者通过在...

    观察者模式for android demo

    5. **EventBus**:另一种在Android中实现观察者模式的方式是使用第三方库EventBus,它可以替代基于BroadcastReceiver的事件传递,简化事件发布和订阅的过程。 6. **回调函数**:在Android开发中,我们也经常使用回...

    C#版本EventBus事件总线实例源码

    在.NET开发环境中,EventBus是一种...通过理解观察者模式和EventBus的核心概念,可以轻松地在项目中实现和使用EventBus,提高代码的可维护性和灵活性。在实际应用中,可以根据需求进行定制和扩展,以满足各种复杂场景。

    android eventbus source code

    我们通常使用观察者模式,正式因为观察者模式非常常见,所以在jdk1.5中已经帮助我们实现了观察者模式,我们只需要简单的继承一些类就可以快速使用观察者模式,在Android中也有一个类似功能的开源库EventBus,...

    EventBus消息总线

    EventBus是Guava的事件处理机制,是设计模式中的观察者模式(生产/消费者编程模型)的优雅实现。对于事件监听和发布订阅模式,EventBus是一个非常优雅和简单解决方案,我们不用创建复杂的类和接口层次结构。 ...

    ,采用MVP开发模式,EventBus数据分发,沉浸式状态栏,ORM,网络请求,数据解析,依赖注入

    本项目是Android快速开发... 预想集成工具包,采用MVP开发模式,EventBus数据分发,沉浸式状态栏,ORM,网络请求(HTTPClint,Volley,OkHttps),数据解析,依赖注入(AndroidAnnotations),xutils,图片异步加载,二维码扫描等等,

    简单EventBus事件总线源码(.net)适合初学者.rar

    EventBus是一种设计模式,常用于实现应用程序内部组件之间的松耦合通信。在.NET框架中,EventBus的概念可以被实现为一个简单的类库,允许不同组件订阅和发布事件,而无需直接依赖于彼此。这个"简单EventBus事件总线...

    EventBus源码及demo

    - **解耦**:EventBus 使得发布者和订阅者之间不需要直接依赖,提高了代码的可维护性。 - **灵活性**:事件可以跨活动、服务、碎片等组件传递,不受组件层级限制。 - **性能优化**:EventBus 可以根据线程模式选择...

    Node.js-Eventbus3.0简单使用适合初学者使用

    **Node.js-Eventbus3.0简单使用适合初学者使用** 在Node.js开发过程中,有时候我们需要在不同的模块之间传递事件,实现解耦合和高效通信。这就是EventBus(事件总线)的概念,它模仿了Android中的EventBus库,提供...

    EventBus范例

    EventBus 使用了观察者模式,允许订阅者注册到事件总线上,当发布者发送事件时,EventBus会根据事件类型找到所有订阅该类型的订阅者,并将事件分发给它们。这个过程是异步的,提高了应用的响应速度。 **使用步骤** ...

    JAVA 发布-订阅(观察者)模式

    发布-订阅(观察者)模式是软件设计模式中的一种行为模式,它在Java中得到了广泛的应用。这种模式的核心思想是定义对象之间的依赖关系,当一个对象的状态发生改变时,所有依赖于它的对象都会得到通知并自动更新。在...

    EventBus 原理简化分析版

    7. **事件分发(Event Delivery)**:EventBus 使用了观察者模式,当一个事件被发布时,它会遍历所有已注册的订阅者并根据线程模型和优先级来分发事件。 8. **异常处理(Exception Handling)**:如果在事件处理...

    eventBus 3.0 带有注释的源码

    在 org 文件夹中,可能包含了 EventBus 的核心类和接口,如 EventBus 类本身、SubscriberMethodFinder 用于查找订阅者方法、Subscription 和 SubscriberInfo 管理订阅者信息、以及相关的辅助工具类。通过对这些源码...

    EventBus Demo

    1. **降低耦合度**:通过发布/订阅模式,EventBus 解耦了发送事件(发布)和接收事件(订阅)的组件,使得它们之间无需直接引用。 2. **简单易用**:事件处理只需简单的注解,无需创建接口或实现类。 3. **高性能**...

Global site tag (gtag.js) - Google Analytics