通过上面的两个示例,你应该对Mina 如何编写TCP/IP 协议栈的网络通信有了一些感性的认识。
(1.)IoService:
这个接口是服务端IoAcceptor、客户端IoConnector 的抽象,提供IO 服务和管理IoSession的功能,它有如下几个常用的方法:
用的方法:
A. TransportMetadata getTransportMetadata():
这个方法获取传输方式的元数据描述信息,也就是底层到底基于什么的实现,譬如:nio、apr 等
B. void addListener(IoServiceListener listener):
这个方法可以为IoService 增加一个监听器,用于监听IoService 的创建、活动、失效、空闲、销毁,具体可以参考IoServiceListener 接口中的方法,这为你参与IoService 的生命周期提供了机会。
C. void removeListener(IoServiceListener listener):
这个方法用于移除上面的方法添加的监听器。
D. void setHandler(IoHandler handler):
这个方法用于向IoService 注册IoHandler,同时有getHandler()方法获取Handler。
E. Map<Long,IoSession> getManagedSessions():
这个方法获取IoService 上管理的所有IoSession,Map 的key 是IoSession 的id。
F. IoSessionConfig getSessionConfig():
这个方法用于获取IoSession 的配置对象,通过IoSessionConfig 对象可以设置Socket 连接的一些选项。
(2.)IoAcceptor:
这个接口是TCPServer 的接口,主要增加了void bind()监听端口、void unbind()解除对套接字的监听等方法。这里与传统的JAVA 中的ServerSocket 不同的是IoAcceptor 可以多次调用bind()方法(或者在一个方法中传入多个SocketAddress 参数)同时监听多个端口。
3.)IoConnector:
这个接口是TCPClient 的接口, 主要增加了ConnectFuture connect(SocketAddressremoteAddress,SocketAddress localAddress)方法,用于与Server 端建立连接,第二个参数如果不传递则使用本地的一个随机端口访问Server 端。这个方法是异步执行的,同样的,也可以同时连接多个服务端。
(4.)IoSession:
这个接口用于表示Server 端与Client 端的连接,IoAcceptor.accept()的时候返回实例。
这个接口有如下常用的方法:
A. WriteFuture write(Object message):
这个方法用于写数据,该操作是异步的。
B. CloseFuture close(boolean immediately):
这个方法用于关闭IoSession,该操作也是异步的,参数指定true 表示立即关闭,否则就在所有的写操作都flush 之后再关闭。
C. Object setAttribute(Object key,Object value):
这个方法用于给我们向会话中添加一些属性,这样可以在会话过程中都可以使用,类似于HttpSession 的setAttrbute()方法。IoSession 内部使用同步的HashMap 存储你添加的自定义属性。
D. SocketAddress getRemoteAddress():
这个方法获取远端连接的套接字地址。
E. void suspendWrite():
这个方法用于挂起写操作,那么有void resumeWrite()方法与之配对。对于read()方法同样适用。
F. ReadFuture read():
这个方法用于读取数据, 但默认是不能使用的, 你需要调用IoSessionConfig 的setUseReadOperation(true)才可以使用这个异步读取的方法。一般我们不会用到这个方法,因为这个方法的内部实现是将数据保存到一个BlockingQueue,假如是Server 端,因为大量的Client 端发送的数据在Server 端都这么读取,那么可能会导致内存泄漏,但对于Client,可能有的时候会比较便利。
G. IoService getService():
这个方法返回与当前会话对象关联的IoService 实例。
(5.)IoSessionConfig:
这个方法用于指定此次会话的配置,它有如下常用的方法:
A. void setReadBufferSize(int size):
这个方法设置读取缓冲的字节数,但一般不需要调用这个方法,因为IoProcessor 会自动调整缓冲的大小。你可以调用setMinReadBufferSize()、setMaxReadBufferSize()方法,这样无论IoProcessor 无论如何自动调整,都会在你指定的区间。
B. void setIdleTime(IdleStatus status,int idleTime):
这个方法设置关联在通道上的读、写或者是读写事件在指定时间内未发生,该通道就进入空闲状态。一旦调用这个方法,则每隔idleTime 都会回调过滤器、IoHandler 中的sessionIdle()方法。
C. void setWriteTimeout(int time):
这个方法设置写操作的超时时间。
D. void setUseReadOperation(boolean useReadOperation):
这个方法设置IoSession 的read()方法是否可用,默认是false。
(6.)IoHandler:
这个接口是你编写业务逻辑的地方,从上面的示例代码可以看出,读取数据、发送数据基本都在这个接口总完成,这个实例是绑定到IoService 上的,有且只有一个实例(没有给一个IoService 注入一个IoHandler 实例会抛出异常)。它有如下几个方法:
A. void sessionCreated(IoSession session):
这个方法当一个Session 对象被创建的时候被调用。对于TCP 连接来说,连接被接受的时候调用,但要注意此时TCP 连接并未建立,此方法仅代表字面含义,也就是连接的对象IoSession 被创建完毕的时候,回调这个方法。对于UDP 来说,当有数据包收到的时候回调这个方法,因为UDP 是无连接的。
B. void sessionOpened(IoSession session):
这个方法在连接被打开时调用,它总是在sessionCreated()方法之后被调用。对于TCP 来说,它是在连接被建立之后调用,你可以在这里执行一些认证操作、发送数据等。对于UDP 来说,这个方法与sessionCreated()没什么区别,但是紧跟其后执行。如果你每隔一段时间,发送一些数据,那么sessionCreated()方法只会在第一次调用,但是sessionOpened()方法每次都会调用。
C. void sessionClosed(IoSession session) :
对于TCP 来说,连接被关闭时,调用这个方法。对于UDP 来说,IoSession 的close()方法被调用时才会毁掉这个方法。
(7.)IoBuffer:
这个接口是对JAVA NIO 的ByteBuffer 的封装,这主要是因为ByteBuffer 只提供了对基本数据类型的读写操作,没有提供对字符串等对象类型的读写方法,使用起来更为方便,另外,ByteBuffer 是定长的,如果想要可变,将很麻烦。IoBuffer 的可变长度的实现类似于StringBuffer。IoBuffer 与ByteBuffer 一样,都是非线程安全的。本节的一些内容如果不清楚,可以参考java.nio.ByteBuffer 接口。这个接口有如下常用的方法:
D. void sessionIdle(IoSession session, IdleStatus status) :
这个方法在IoSession 的通道进入空闲状态时调用,对于UDP 协议来说,这个方法始终不会被调用。
E. void exceptionCaught(IoSession session, Throwable cause) :
这个方法在你的程序、Mina 自身出现异常时回调,一般这里是关闭IoSession。
F. void messageReceived(IoSession session, Object message) :
接收到消息时调用的方法,也就是用于接收消息的方法,一般情况下,message 是一个IoBuffer 类,如果你使用了协议编解码器,那么可以强制转换为你需要的类型。通常我们都是会使用协议编解码器的, 就像上面的例子, 因为协议编解码器是
TextLineCodecFactory,所以我们可以强制转message 为String 类型。
G. void messageSent(IoSession session, Object message) :
当发送消息成功时调用这个方法,注意这里的措辞,发送成功之后,也就是说发送消息是不能用这个方法的。
发送消息的时机:
发送消息应该在sessionOpened()、messageReceived()方法中调用IoSession.write()方法完成。因为在sessionOpened()方法中,TCP 连接已经真正打开,同样的在messageReceived()方法TCP 连接也是打开状态,只不过两者的时机不同。sessionOpened()方法是在TCP 连接建立之后,接收到数据之前发送;messageReceived()方法是在接收到数据之后发送,你可以完成依据收到的内容是什么样子,决定发送什么样的数据。因为这个接口中的方法太多,因此通常使用适配器模式IoHandlerAdapter,覆盖你所感兴趣的方法即可。
A. static IoBuffer allocate(int capacity,boolean useDirectBuffer):
这个方法内部通过SimpleBufferAllocator 创建一个实例,第一个参数指定初始化容量,第二个参数指定使用直接缓冲区还是JAVA 内存堆的缓存区,默认为false。
B. void free():
释放缓冲区,以便被一些IoBufferAllocator 的实现重用,一般没有必要调用这个方法,除非你想提升性能(但可能未必效果明显)。
C. IoBuffer setAutoExpand(boolean autoExpand):
这个方法设置IoBuffer 为自动扩展容量,也就是前面所说的长度可变,那么可以看出长度可变这个特性默认是不开启的。
D. IoBuffer setAutoShrink(boolean autoShrink):
这个方法设置IoBuffer 为自动收缩,这样在compact()方法调用之后,可以裁减掉一些没有使用的空间。如果这个方法没有被调用或者设置为false,你也可以通过调用shrink()方法手动收缩空间。
E. IoBuffer order(ByteOrder bo):
这个方法设置是Big Endian 还是Little Endian,JAVA 中默认是Big Endian,C++和其他语言一般是Little Endian。
F. IoBuffer asReadOnlyBuffer():
这个方法设置IoBuffer 为只读的。
G. Boolean prefixedDataAvailable(int prefixLength,int maxDataLength):
这个方法用于数据的最开始的1、2、4 个字节表示的是数据的长度的情况,
prefixLentgh表示这段数据的前几个字节(只能是1、2、4 的其中一个),代表的是这段数据的长度,
maxDataLength 表示最多要读取的字节数。返回结果依赖于等式
remaining()-prefixLength>=maxDataLength,也就是总的数据-表示长度的字节,剩下的字节数要比打算读取的字节数大或者相等。
H. String getPrefixedString(int prefixLength,CharsetDecoder decoder):
如果上面的方法返回true,那么这个方法将开始读取表示长度的字节之后的数据,注意要保持这两个方法的prefixLength 的值是一样的。
G、H 两个方法在后面讲到的PrefixedStringDecoder 中的内部实现使用。
IoBuffer 剩余的方法与ByteBuffer 都是差不多的,额外增加了一些便利的操作方法,例如:
IoBuffer putString(String value,CharsetEncoder encoder)可以方便的以指定的编码方式存储字符串、InputStream asInputStream()方法从IoBuffer 剩余的未读的数据中转为输入流等。
(8.)IoFuture:
在Mina 的很多操作中,你会看到返回值是XXXFuture,实际上他们都是IoFuture 的子类,看到这样的返回值,这个方法就说明是异步执行的,主要的子类有ConnectFuture、CloseFuture 、ReadFuture 、WriteFuture 。这个接口的大部分操作都和
java.util.concurrent.Future 接口是类似的,譬如:await()、awaitUninterruptibly()等,一般我们常用awaitUninterruptibly()方法可以等待异步执行的结果返回。这个接口有如下常用的方法:
A. IoFuture addListener(IoFutureListener<?> listener):
这个方法用于添加一个监听器, 在异步执行的结果返回时监听器中的回调方法operationComplete(IoFuture future),也就是说,这是替代awaitUninterruptibly()方法另一种等待异步执行结果的方法,它的好处是不会产生阻塞。
B. IoFuture removeListener(IoFutureListener<?> listener):
这个方法用于移除指定的监听器。
C. IoSession getSession():
这个方法返回当前的IoSession。举个例子,我们在客户端调用connect()方法访问Server 端的时候,实际上这就是一个异步执行的方法,也就是调用connect()方法之后立即返回,执行下面的代码,而不管是否连接成功。
那么如果我想在连接成功之后执行一些事情(譬如:获取连接成功后的IoSession对象),该怎么办呢?按照上面的说明,你有如下两种办法:
第一种:
第二种:
为了更好的看清楚使用监听器是异步的,而不是像awaitUninterruptibly()那样会阻塞主线程的执行,我们在回调方法中暂停5 秒钟,然后输出+++,在最后输出***。我们执行代码之后,你会发现首先输出***(这证明了监听器是异步执行的),然后IoSession 对象Created,系统暂停5 秒,然后输出+++,最后IoSession 对象Opened,也就是TCP 连接建立。
4.日志配置:
前面的示例代码中提到了使用SLF4J 作为日志门面,这是因为Mina 内部使用的就是SLF4J,你也使用SLF4J 可以与之保持一致性。Mina 如果想启用日志跟踪Mina 的运行细节,你可以配置LoggingFilter 过滤器,这样你可
以看到Session 建立、打开、空闲等一系列细节在日志中输出,默认SJF4J 是按照DEBUG级别输出跟踪信息的,如果你想给某一类别的Mina 运行信息输出指定日志输出级别,可以调用LoggingFilter 的setXXXLogLevel(LogLevel.XXX)。
这里IoSession 被打开的跟踪信息将以ERROR 级别输出到日志。
5.过滤器:
前面我们看到了LoggingFilter、ProtocolCodecFilter 两个过滤器,一个负责日志输出,一个负责数据的编解码,通过最前面的Mina 执行流程图,在IoProcessor 与IoHandler 之间可以有很多的过滤器,这种设计方式为你提供可插拔似的扩展功能提供了非常便利的方式,目前的Apache CXF、Apache Struts2 中的拦截器也都是一样的设计思路。Mina 中的IoFilter 是单例的,这与CXF、Apache Struts2 没什么区别。IoService 实例上会绑定一个DefaultIoFilterChainBuilder 实例,DefaultIoFilterChainBuilder 会把使用内部的EntryImpl 类把所有的过滤器按照顺序连在一起,组成一个过滤器链。
DefaultIoFilterChainBuilder 类如下常用的方法:
A. void addFirst(String name,IoFilter filter):
这个方法把过滤器添加到过滤器链的头部,头部就是IoProcessor 之后的第一个过滤器。同样的addLast()方法把过滤器添加到过滤器链的尾部。
B. void addBefore(String baseName,String name,IoFilter filter):
这个方法将过滤器添加到baseName 指定的过滤器的前面,同样的addAfter()方法把过滤器添加到baseName 指定的过滤器的后面。这里要注意无论是那种添加方法,每个过滤器的名字(参数name)必须是唯一的。
C. IoFilter remove(Stirng name):
这个方法移除指定名称的过滤器,你也可以调用另一个重载的remove()方法,指定要移除的IoFilter 的类型。
这个方法返回当前IoService 上注册的所有过滤器。默认情况下,过滤器链中是空的,也就是getAll()方法返回长度为0 的List,但实际Mina内部有两个隐藏的过滤器:HeadFilter、TailFilter,分别在List 的最开始和最末端,很明显,TailFilter 在最末端是为了调用过滤器链之后,调用IoHandler。但这两个过滤器对你来说是透明的,可以忽略它们的存在。编写一个过滤器很简单,你需要实现IoFilter 接口,如果你只关注某几个方法,可以继承IoFilterAdapter 适配器类。IoFilter 接口中主要包含两类方法,一类是与IoHandler 中的方法名一致的方法,相当于拦截IoHandler 中的方法,另一类是IoFilter 的生命周期回调方法,这些回调方法的执行顺序和解释如下所示:
(1.)init()在首次添加到链中的时候被调用,但你必须将这个IoFilter 用
ReferenceCountingFilter 包装起来,否则init()方法永远不会被调用。
(2.)onPreAdd()在调用添加到链中的方法时被调用,但此时还未真正的加入到链。
(3.)onPostAdd()在调用添加到链中的方法后被调,如果在这个方法中有异常抛出,则过滤器会立即被移除,同时destroy()方法也会被调用(前提是使用ReferenceCountingFilter包装)。
(4.)onPreRemove()在从链中移除之前调用。
(5.)onPostRemove()在从链中移除之后调用。
(6.)destory()在从链中移除时被调用,使用方法与init()要求相同。
无论是哪个方法,要注意必须在实现时调用参数nextFilter 的同名方法,否则,过滤器链的执行将被中断,IoHandler 中的同名方法一样也不会被执行,这就相当于Servlet 中的Filter 必须调用filterChain.doFilter(request,response)才能继续前进是一样的道理。
这里我们将MyIoFilter 用ReferenceCountingFilter 包装起来,这样你可以看到init()、destroy()方法调用。我们启动客户端访问,然后关闭客户端,你会看到执行顺序如下所示:
init onPreAdd onPostAdd sessionCreated sessionOpened messageReceived filterClose sessionClosed onPreRemove onPostRemove destroy。
IoHandler 的对应方法会跟在上面的对应方法之后执行,这也就是说从横向(单独的看一个过滤器中的所有方法的执行顺序)上看,每个过滤器的执行顺序是上面所示的顺序;从纵向(方法链的调用)上看,如果有filter1、filter2 两个过滤器,sessionCreated()方法的执行顺序如下所示:
filter1-sessionCreated filter2-sessionCreated IoHandler-sessionCreated。
这里你要注意init、onPreAdd、onPostAdd 三个方法并不是在Server 启动时调用的,而是IoSession 对象创建之前调用的,也就是说IoFilterChain.addXXX()方法仅仅负责初始化过滤器并注册过滤器,但并不调用任何方法,包括init()初始化方法也是在IoProcessor 开始工作的时候被调用。IoFilter 是单例的,那么init()方法是否只被执行一次呢?这个是不一定的,因为IoFilter是被IoProcessor 调用的,而每个IoService 通常是关联多个IoProcessor,所以IoFilter的init()方法是在每个IoProcessor 线程上只执行一次。关于Mina 的线程问题,我们后面会详细讨论,这里你只需要清楚,init()与destroy()的调用次数与IoProceesor 的个数有关,假如一个IoService 关联了3 个IoProcessor,有五个并发的客户端请求,那么你会看到三次init()方法被调用,以后将不再会调用。Mina中自带的过滤器:
过滤器 说明
BlacklistFilter 设置一些IP 地址为黑名单,不允许访问。
BufferedWriteFilter 设置输出时像BufferedOutputStream 一样进行缓冲。
CompressionFilter 设置在输入、输出流时启用JZlib 压缩。
ConnectionThrottleFilter 这个过滤器指定同一个IP 地址(不含端口号)上的请求在多长的毫秒值内可以有一个请求,如果小于指定的时间间隔就有连续两个请求,那么第二个请求将被忽略(IoSession.close())。正如Throttle 的名字一样,调节访问的频率这个过滤器最好放在过滤器链的前面。
FileRegionWriteFilter 如果你想使用File 对象进行输出,请使用这个过滤器。要注意,你需要使用WriteFuture 或者在
messageSent() 方法中关闭File 所关联的FileChannel 通道。
StreamWriteFilter 如果你想使用InputStream 对象进行输出,请使用这个过滤器。要注意,你需要使用WriteFuture或者在messageSent()方法中关闭File 所关联的
FileChannel 通道。NoopFilter 这个过滤器什么也不做,如果你想测试过滤器链是否起作用,可以用它来测试。
ProfilerTimerFilter 这个过滤器用于检测每个事件方法执行的时间,所以最好放在过滤器链的前面。
ProxyFilter 这个过滤器在客户端使用ProxyConnector 作为实现时,会自动加入到过滤器链中,用于完成代理功能。
RequestResponseFilter 暂不知晓。
SessionAttributeInitializingFilter 这个过滤器在IoSession 中放入一些属性(Map),通常放在过滤器的前面,用于放置一些初始化的信息。
MdcInjectionFilter 针对日志输出做MDC 操作,可以参考LOG4J 的MDC、NDC 的文档。
WriteRequestFilter CompressionFilter、RequestResponseFilter 的基类,用于包装写请求的过滤器。
还有一些过滤器,会在各节中详细讨论,这里没有列出,譬如:前面的LoggingFilger 日志过滤器。