分享

深入解析Apache Mina源码(2)——Mina的事件模型

 集微笔记 2014-02-24
一、观察者模式的本来面目

工作时间长了,会发现代码中的很多东西都是相通相似的,就说JAVA的事件机制其实就是观察者模式的实现,学会了观察者模式,事件机制自己无师自通,先以我的角度看看什么是观察者模式。

观察者模式,顾名思义,应该有观察者(抽象观察者“Observer”角色)和被观察者(抽象主题“Subject”角色),被观察者一旦有变化就通知观察者更新自己。

形象图:



下面是观察者模式的类图:




从我的角度简单点理解,我们知道面向对象编程不过是对象之间的相互调用,那么被观察者应该有个列表(List)来把所有观察者引用起来,当有事件要通知观察者时,我们从List中把所有观察者类取出来挨个调用观察者的方法(如update())。这个过程中只有充分利用了面向对象编程的多态,继承等特性才能实现。

下面是代码实现:

被观察者(抽象主题类):

Java代码 复制代码 收藏代码
  1. package com.lifanghu.observer;   
  2.   
  3. import java.util.ArrayList;   
  4. import java.util.List;   
  5.   
  6. /**  
  7.  * @author lifh  
  8.  * @mail wslfh2005@163.com  
  9.  * @since 2012-6-14 下午06:18:26  
  10.  * @name com.lifanghu.observer.Subject.java  
  11.  * @version 1.0  
  12.  */  
  13. public abstract class Subject {   
  14.   
  15.     private List observers = new ArrayList();   
  16.   
  17.     /**  
  18.      * 增加一个观察者  
  19.      * @param observer  
  20.      * @author lifh  
  21.      */  
  22.     public void addObserver(Observer observer) {   
  23.         observers.add(observer);   
  24.     }   
  25.     /**  
  26.      * 删除一个观察者  
  27.      * @param observer  
  28.      * @author lifh  
  29.      */  
  30.     public void removeObserver(Observer observer) {   
  31.         observers.remove(observer);   
  32.     }   
  33.     /**  
  34.      * 通知所有观察者  
  35.      * @author lifh  
  36.      */  
  37.     public void notifyObservers() {   
  38.         for (Observer observer : observers) {   
  39.             observer.update(this);   
  40.         }   
  41.     }   
  42. }   

  实际被观察者(真实主题角色):

Java代码 复制代码 收藏代码
  1. package com.lifanghu.observer;   
  2.   
  3. /**  
  4.  * 实际被观察者  
  5.  * @author lifh  
  6.  * @mail wslfh2005@163.com  
  7.  * @since 2012-6-14 下午10:48:02  
  8.  * @name com.lifanghu.observer.XiaoMing.java  
  9.  * @version 1.0  
  10.  */  
  11. public class XiaoMing extends Subject {   
  12.   
  13.     private String state;   
  14.   
  15.     //实际要通进行通知的方法   
  16.     public void change() {   
  17.         this.notifyObservers();   
  18.     }   
  19.     public String getState() {   
  20.         return state;   
  21.     }   
  22.     public void setState(String state) {   
  23.         this.state = state;   
  24.     }   
  25. }  
 抽象观察者:
Java代码 复制代码 收藏代码
  1. package com.lifanghu.observer;   
  2.   
  3. /**  
  4.  * 实际观察者  
  5.  * @author lifh  
  6.  * @mail wslfh2005@163.com  
  7.  * @since 2012-6-14 下午10:54:07  
  8.  * @name com.lifanghu.observer.XiaoWang.java  
  9.  * @version 1.0  
  10.  */  
  11. public class XiaoWang implements Observer {   
  12.   
  13.     public void update(Subject subject) {   
  14.         XiaoMing xm = (XiaoMing) subject;   
  15.         System.out.println("小王得到通知:" + xm.getState() + ";小王说:活该!");   
  16.     }   
  17. }   

  具体观察者1:

Java代码 复制代码 收藏代码
  1. package com.lifanghu.observer;   
  2.   
  3. /**  
  4.  * 实际观察者  
  5.  * @author lifh  
  6.  * @mail wslfh2005@163.com  
  7.  * @since 2012-6-14 下午10:54:07  
  8.  * @name com.lifanghu.observer.XiaoWang.java  
  9.  * @version 1.0  
  10.  */  
  11. public class XiaoWang implements Observer {   
  12.   
  13.     public void update(Subject subject) {   
  14.         XiaoMing xm = (XiaoMing) subject;   
  15.         System.out.println("小王得到通知:" + xm.getState() + ";小王说:活该!");   
  16.     }   
  17. }  
 具体观察者2:
Java代码 复制代码 收藏代码
  1. package com.lifanghu.observer;   
  2.   
  3. /**  
  4.  * 实际观察者  
  5.  * @author lifh  
  6.  * @mail wslfh2005@163.com  
  7.  * @since 2012-6-14 下午10:56:17  
  8.  * @name com.lifanghu.observer.ZhangShan.java  
  9.  * @version 1.0  
  10.  */  
  11. public class ZhangShan implements Observer {   
  12.   
  13.     public void update(Subject subject) {   
  14.         XiaoMing xm = (XiaoMing) subject;   
  15.         System.out.println("张三得到通知:" + xm.getState() + ";张三说:快吃药吧!");   
  16.     }   
  17. }  
 客户端调用逻辑:
Java代码 复制代码 收藏代码
  1. package com.lifanghu.observer;   
  2.   
  3. /**  
  4.  * 实际观察者  
  5.  * @author lifh  
  6.  * @mail wslfh2005@163.com  
  7.  * @since 2012-6-14 下午10:56:17  
  8.  * @name com.lifanghu.observer.ZhangShan.java  
  9.  * @version 1.0  
  10.  */  
  11. public class ZhangShan implements Observer {   
  12.   
  13.     public void update(Subject subject) {   
  14.         XiaoMing xm = (XiaoMing) subject;   
  15.         System.out.println("张三得到通知:" + xm.getState() + ";张三说:快吃药吧!");   
  16.     }   
  17. }  
 输出结果:
 写道
张三得到通知:小明病了;张三说:快吃药吧!
小王得到通知:小明病了;小王说:活该!
        说明:对于观察者模式JDK是有相应的实现支持的,内置观察者模式主要有2个类,一个是类Observable,一个是接口类Observer ,大家可以上网去查,或者直接看它的源码,这里就不多说了。

 

二、从观察者模式到事件机制

关于事件机制上面也谈了,它其实就是观察者模式的实现,关于两者的联系也可以看iteye上的另一讨论http://www./topic/182643,里面说的已经很清楚了。

关于事件模型我想说的是里面的三个元素:事件源,事件本身,监听者,对应观察者模式的三个元素分别是:被观察者,观察者方法要传入的参数,观察者。

关于事件机制JDK也有相应的实现java.util.EventListenerjava.util.EventObject

三、Mina中的事件以及监听器的实现

Mina 中有主要有三种事件:

1org.apache.mina.core.service包下面对IoService接口实现和IoSession接口实现的监听,它不是一个纯的事件模型,没有事件本身(EventObject)的存在。

2org.apache.mina.core.session包下在线程池环境下对于事件的触发从而进入过滤器链进行的处理,它不是一个纯事件的模型,少了监听器(EventListener)的实现。

3org.apache.mina.core.future包下对于异步IO操作(IoFuture)进行监听。它也不是一个纯的事件模型,也是少了事件本身(EventObject)的存在。

下面分别介绍下这三种事件细节。

1service包下面主要有两个类是用于监听模式的,监听者IoServiceListener和它的帮助类IoServiceListenerSupportIoServiceListener接口继承了EventListener,标明自己是个监听者接口,主要监听servicesession的创建,空闲,销毁事件的。IoServiceListenerSupport作为它的帮助类,实际上充当了事件源的角色,它存储了IoServiceListener的成员列表。

Java代码 复制代码 收藏代码
  1. /** A list of {@link IoServiceListener}s. */  
  2. private final List listeners = new CopyOnWriteArrayList();  

 里面有增加和删除监听者的方法:

Java代码 复制代码 收藏代码
  1. /**  
  2.  * Adds a new listener.  
  3.  *   
  4.  * @param listener The added listener  
  5.  */  
  6. public void add(IoServiceListener listener) {   
  7.     if (listener != null) {   
  8.         listeners.add(listener);   
  9.     }   
  10. }   
  11.   
  12. /**  
  13.  * Removes an existing listener.  
  14.  *   
  15.  * @param listener The listener to remove  
  16.  */  
  17. public void remove(IoServiceListener listener) {   
  18.     if (listener != null) {   
  19.         listeners.remove(listener);   
  20.     }   
  21. }  

通知监听者的方法:

Java代码 复制代码 收藏代码
  1. /**  
  2.  * Calls {@link IoServiceListener#serviceActivated(IoService)}  
  3.  * for all registered listeners.  
  4.  * 通知方法  
  5.  */  
  6. public void fireServiceActivated() {   
  7.     if (!activated.compareAndSet(falsetrue)) {   
  8.         // The instance is already active   
  9.         return;   
  10.     }   
  11.   
  12.     activationTime = System.currentTimeMillis();   
  13.   
  14.     // Activate all the listeners now   
  15.     for (IoServiceListener listener : listeners) {//循环提取出listeners执行本身的方法   
  16.         try {   
  17.             listener.serviceActivated(service);   
  18.         } catch (Throwable e) {   
  19.             ExceptionMonitor.getInstance().exceptionCaught(e);   
  20.         }   
  21.     }   
  22. }  

再来看接口IoService,有对监听器的操作:

Java代码 复制代码 收藏代码
  1. /**  
  2.  * Adds an {@link IoServiceListener} that listens any events related with  
  3.  * this service.  
  4.  */  
  5. void addListener(IoServiceListener listener);   
  6.   
  7. /**  
  8.  * Removed an existing {@link IoServiceListener} that listens any events  
  9.  * related with this service.  
  10.  */  
  11. void removeListener(IoServiceListener listener);  

 它的实现AbstractIoService里面可以看到,实际是调用的IoServiceListenerSupport的方法执行的有关操作:

Java代码 复制代码 收藏代码
  1. /**  
  2.  * {@inheritDoc}  
  3.  */  
  4. public final void addListener(IoServiceListener listener) {   
  5.     listeners.add(listener);   
  6. }   
  7.   
  8. /**  
  9.  * {@inheritDoc}  
  10.  */  
  11. public final void removeListener(IoServiceListener listener) {   
  12.     listeners.remove(listener);   
  13. }  

 看下AbstractIoAcceptor类的调用方法:

Java代码 复制代码 收藏代码
  1. if (activate) {   
  2.     //触发监听器的服务创建事件   
  3.     getListeners().fireServiceActivated();   
  4. }  

 2session包下的事件模型主要还是基于多线程模型来实现的,可以看到IoEvent实现了Runnable接口,它作为任务放入到线程池中去执行,看它的run方法:

Java代码 复制代码 收藏代码
  1. // 此类实现了 Runnable 它的存在主要为了线程池过滤器ExecutorFilter来使用的。   
  2. // 这样如果加入了ExecutorFilter后,后面所有的数据处理都是多线程方式进行的。   
  3. // 一般放在handler前面,也就是所有过滤器的后面,因为handler主要做业务处理,可能会有数据库的操作,比较耗时,适合多线程。   
  4. public void run() {   
  5.     //激活各种事件   
  6.     fire();   
  7. }  

 IO事件类型:

Java代码 复制代码 收藏代码
  1. package org.apache.mina.core.session;   
  2.   
  3. /**  
  4.  * An {@link Enum} that represents the type of I/O events and requests.  
  5.  * Most users won't need to use this class.  It is usually used by internal  
  6.  * components to store I/O events.  
  7.  *  
  8.  * @author Apache MINA Project  
  9.  */  
  10. public enum IoEventType {   
  11.     SESSION_CREATED,//session创建   
  12.     SESSION_OPENED,//session打开   
  13.     SESSION_CLOSED,//session关闭   
  14.     MESSAGE_RECEIVED,//消息接收   
  15.     MESSAGE_SENT,//消息发送   
  16.     SESSION_IDLE,//session空闲   
  17.     EXCEPTION_CAUGHT,//发生异常   
  18.     WRITE,//写事件   
  19.     CLOSE,//关闭session事件   
  20. }  

 它有一个实现类IoFilterEvent,我们知道如果我们想要利用mina的多线程处理,需要如下加一个过滤器:

Java代码 复制代码 收藏代码
  1. connector.getFilterChain().addLast("executor"new ExecutorFilter());  

 在这个时候我们的IoEvent就派上了用场,ExecutorFilter是一个实现了线程池(线程池在后面的文章中会介绍)的过滤器,看下ExecutorFilter的调用:

Java代码 复制代码 收藏代码
  1. @Override  
  2. public final void sessionOpened(NextFilter nextFilter, IoSession session) {   
  3.     if (eventTypes.contains(IoEventType.SESSION_OPENED)) {   
  4.         IoFilterEvent event = new IoFilterEvent(nextFilter, IoEventType.SESSION_OPENED,   
  5.             session, null);    
  6.         //触发事件   
  7.         fireEvent(event);   
  8.     } else {   
  9.         nextFilter.sessionOpened(session);   
  10.     }   
  11. }   
  12. /**  
  13.  * Fires the specified event through the underlying executor.  
  14.  *   
  15.  * @param event The filtered event  
  16.  */  
  17. protected void fireEvent(IoFilterEvent event) {   
  18.     //将事件提交给线程池执行   
  19.     executor.execute(event);   
  20. }  

 可以看到这个session下的事件模型没有监听器,所以基本就没有事件模式的概念了,但是还是有一些事件模型的一些影子,所以可以认为它是一个伪事件模型。

3future包下面的事件模型,是比较常规的事件模型,这是一个异步调用后的事件模型,当设定的异步操作完成后会调用监听器的operationComplete方法,事件源为IoFuture等,监听器是IoFutureListener,它只有一个方法:

Java代码 复制代码 收藏代码
  1. /**  
  2.  * Invoked when the operation associated with the {@link IoFuture}  
  3.  * has been completed even if you add the listener after the completion.  
  4.  * 只有一个处理完成的回调方法,当线程池将我们的任务处理完成后会调用此方法。  
  5.  * @param future  The source {@link IoFuture} which called this  
  6.  *                callback.  
  7.  */  
  8. void operationComplete(F future);  

 看下事件源的代码:

Java代码 复制代码 收藏代码
  1. package org.apache.mina.core.future;   
  2.   
  3. import java.util.concurrent.TimeUnit;   
  4.   
  5. import org.apache.mina.core.session.IoSession;   
  6.   
  7. /**  
  8.  * Represents the completion of an asynchronous I/O operation on an   
  9.  * {@link IoSession}.  
  10.  * Can be listened for completion using a {@link IoFutureListener}.  
  11.  * 因为有了线程池模型才会有异步的概念。  
  12.  * 我们将各种IO操作(连接,关闭,读,写)以任务的方式放入队列中并返回IoFuture供线程池去处理。  
  13.  * 处理过程和处理完成的操作会改变IoFuture的状态。  
  14.  * @author Apache MINA Project  
  15.  */  
  16. public interface IoFuture {   
  17.     /**  
  18.      * Returns the {@link IoSession} which is associated with this future.  
  19.      */  
  20.     IoSession getSession();   
  21.   
  22.     /**  
  23.      * Wait for the asynchronous operation to complete.  
  24.      * The attached listeners will be notified when the operation is   
  25.      * completed.  
  26.      */  
  27.     IoFuture await() throws InterruptedException;   
  28.   
  29.     /**  
  30.      * Wait for the asynchronous operation to complete with the specified timeout.  
  31.      * 等待操作在指定时间内完成  
  32.      * @return true if the operation is completed.  
  33.      */  
  34.     boolean await(long timeout, TimeUnit unit) throws InterruptedException;   
  35.   
  36.     /**  
  37.      * Wait for the asynchronous operation to complete with the specified timeout.  
  38.      *  
  39.      * @return true if the operation is completed.  
  40.      */  
  41.     boolean await(long timeoutMillis) throws InterruptedException;   
  42.   
  43.     /**  
  44.      * Wait for the asynchronous operation to complete uninterruptibly.  
  45.      * The attached listeners will be notified when the operation is   
  46.      * completed.  
  47.      * 等待异步操作完成,完成后会触发注册的监听器。  
  48.      * @return the current IoFuture  
  49.      */  
  50.     IoFuture awaitUninterruptibly();   
  51.   
  52.     /**  
  53.      * Wait for the asynchronous operation to complete with the specified timeout  
  54.      * uninterruptibly.  
  55.      *  
  56.      * @return true if the operation is completed.  
  57.      */  
  58.     boolean awaitUninterruptibly(long timeout, TimeUnit unit);   
  59.   
  60.     /**  
  61.      * Wait for the asynchronous operation to complete with the specified timeout  
  62.      * uninterruptibly.  
  63.      *  
  64.      * @return true if the operation is finished.  
  65.      */  
  66.     boolean awaitUninterruptibly(long timeoutMillis);   
  67.   
  68.     /**  
  69.      * @deprecated Replaced with {@link #awaitUninterruptibly()}.  
  70.      */  
  71.     @Deprecated  
  72.     void join();   
  73.   
  74.     /**  
  75.      * @deprecated Replaced with {@link #awaitUninterruptibly(long)}.  
  76.      */  
  77.     @Deprecated  
  78.     boolean join(long timeoutMillis);   
  79.   
  80.     /**  
  81.      * Returns if the asynchronous operation is completed.  
  82.      */  
  83.     boolean isDone();   
  84.   
  85.     /**  
  86.      * Adds an event listener which is notified when  
  87.      * this future is completed. If the listener is added  
  88.      * after the completion, the listener is directly notified.  
  89.      * 增加异步完成后的监听者  
  90.      */  
  91.     IoFuture addListener(IoFutureListener listener);   
  92.   
  93.     /**  
  94.      * Removes an existing event listener so it won't be notified when  
  95.      * the future is completed.  
  96.      * 删除异步完成后的监听者  
  97.      */  
  98.     IoFuture removeListener(IoFutureListener listener);   
  99. }  

 看下调用监听者的方法类DefaultIoFuture的代码:

Java代码 复制代码 收藏代码
  1. /**  
  2.  * Sets the result of the asynchronous operation, and mark it as finished.  
  3.  */  
  4. public void setValue(Object newValue) {   
  5.     synchronized (lock) {   
  6.         // Allow only once.   
  7.         if (ready) {   
  8.             return;   
  9.         }   
  10.   
  11.         result = newValue;   
  12.         ready = true;   
  13.         if (waiters > 0) {   
  14.             lock.notifyAll();   
  15.         }   
  16.     }   
  17.   
  18.     //调用监听者的方法   
  19.     notifyListeners();   
  20. }  

 在此我们可以学到JAVA的异步只有在多线程操作下才有可能实现。

四、推荐文章

参考文章:

1. 观察者模式

http://ttitfly./blog/152512

2. JAVA与模式》之观察者模式

http://www.cnblogs.com/java-my-life/archive/2012/05/16/2502279.html

3. Java事件机制理解及应用

http://blog.csdn.net/JianZhiZG/article/details/1427073

4. java 事件机制

http://www./chenweicai/archive/2007/04/13/110350.html

五、总结

事件模型作为mina中一个重要的设计模式很好的体现了JAVA高内聚,低耦合的设计思想,也让用户的调用代码简洁易用,本文章本着抛砖引玉的态度希望大家能指出缺点,共同讨论。

    本站是提供个人知识管理的网络存储空间,所有内容均由用户发布,不代表本站观点。请注意甄别内容中的联系方式、诱导购买等信息,谨防诈骗。如发现有害或侵权内容,请点击一键举报。
    转藏 分享 献花(0

    0条评论

    发表

    请遵守用户 评论公约

    类似文章 更多