1.MINA框架简介
MINA(Multipurpose Infrastructure for Network Applications)是用于开发高性能和高可用性的网络应用程序的基础框架。通过使用MINA框架可以可以省下处理底层I/O和线程并发等复杂工作,开发人员能够把更多的精力投入到业务设计和开发当中。MINA框架的应用比较广泛,应用的开源项目有Apache Directory、AsyncWeb、Apache Qpid、QuickFIX/J、Openfire、SubEthaSTMP、red5等。MINA框架当前稳定版本是1.1.6,最新的2.0版本目前已经发布了M1版本。
MINA框架的特点有:基于java NIO类库开发;采用非阻塞方式的异步传输;事件驱动;支持批量数据传输;支持TCP、UDP协议;控制反转的设计模式(支持Spring);采用优雅的松耦合架构;可灵活的加载过滤器机制;单元测试更容易实现;可自定义线程的数量,以提高运行于多处理器上的性能;采用回调的方式完成调用,线程的使用更容易。
2.MINA框架的常用类
类NioSocketAcceptor用于创建服务端监听;
类NioSocketConnector用于创建客户端连接;
类IoSession用来保存会话属性和发送消息;
类IoHandlerAdapter用于定义业务逻辑,常用的方法有:
方法 定义
sessionCreated() 当会话创建时被触发
sessionOpened() 当会话开始时被触发
sessionClosed() 当会话关闭时被触发
sessionIdle() 当会话空闲时被触发
exceptionCaught() 当接口中其他方法抛出异常未被捕获时触发此方法
messageRecieved() 当接收到消息后被触发
messageSent() 当发送消息后被触发
3.服务端应用开发示例
下面将以MINA2.0M1版本为基础,通过一个范例来演示一下如何使用MINA框架进行开发。开发环境为jdk6.0,开发工具NetBeans6.0,所需jar包slf4j-api.jar、slf4j-jdk14.jar、MINA-core-2.0.0-M1.jar。
首先定义一个业务逻辑处理器TimeServerHandler,继承自IoHandlerAdapter,实现的功能有:当客户端创建会话时会显示客户端设备的IP和端口;当客户端输入quit时结束会话;客户端输入其它内容时则向客户端发送当前时间。代码如下:
1 public class TimeServerHandler extends IoHandlerAdapter 2 { 3 @Override 4 public void sessionCreated(IoSession session) { 5 //显示客户端的ip和端口 6 System.out.println(session.getRemoteAddress().toString()); 7 } 8 @Override 9 public void messageReceived( IoSession session, Object message ) throws Exception 10 { 11 String str = message.toString(); 12 if( str.trim().equalsIgnoreCase("quit") ) { 13 session.close();//结束会话 14 return; 15 } 16 Date date = new Date(); 17 session.write( date.toString() );//返回当前时间的字符串 18 System.out.println("Message written..."); 19 } 20 } 21 再定义一个类MinaTimeServer用来启动服务端: 22 public class MinaTimeServer 23 { 24 private static final int PORT = 9123;//定义监听端口 25 public static void main( String[] args ) throws IOException 26 { 27 IoAcceptor acceptor = new NioSocketAcceptor(); 28 acceptor.getFilterChain().addLast( "logger", new LoggingFilter() ); 29 acceptor.getFilterChain().addLast( "codec", new ProtocolCodecFilter( new TextLineCodecFactory( Charset.forName( "UTF-8" ))));//指定编码过滤器 30 acceptor.setHandler( new TimeServerHandler() );//指定业务逻辑处理器 31 acceptor.setDefaultLocalAddress( new InetSocketAddress(PORT) );//设置端口号 32 acceptor.bind();//启动监听 33 } 34 }
4.测试
首先运行MinaTimeServer,启动服务端,接着在命令行运行“telnet 127.0.0.1 9123”,来登录,这时会看到服务端输出如下:
2008-2-21 16:15:29 org.apache.MINA.filter.logging.LogLevel$4 log
/10.64.2.137:4140 ?IP和端口号
信息: CREATED
2008-2-21 16:15:29 org.apache.MINA.filter.logging.LogLevel$4 log
信息: OPENED 在客户端输入回车,在客户端可以看到服务端返回当前的时间:
Thu Feb 21 16:20:14 CST 2008
服务端输出:
2008-2-21 16:20:14 org.apache.MINA.filter.logging.LogLevel$4 log
信息: RECEIVED: HeapBuffer[pos=0 lim=2 cap=2048: 0D 0A] ?接收收到回车符
Message written...
2008-2-21 16:20:14 org.apache.MINA.filter.logging.LogLevel$4 log
信息: SENT: HeapBuffer[pos=0 lim=29 cap=30: 54 68 75 20 46 65 62 20 32 31 20 31 36 3A 32 30...]
2008-2-21 16:20:14 org.apache.MINA.filter.logging.LogLevel$4 log
信息: SENT: HeapBuffer[pos=0 lim=0 cap=0: empty] 5.客户端开发示例
首先定义类TimeClientHandler来处理消息接收事件:
1 class TimeClientHandler extends IoHandlerAdapter{ 2 public TimeClientHandler() { 3 } 4 @Override 5 public void messageReceived(IoSession session, Object message) throws Exception { 6 System.out.println(message);//显示接收到的消息 7 } 8 } 9 接着定义MinaTimeClient类用于连接服务端,并向服务端发送消息: 10 public class MinaTimeClient { 11 public static void main(String[] args) { 12 // 创建客户端连接器. 13 NioSocketConnector connector = new NioSocketConnector(); 14 connector.getFilterChain().addLast( "logger", new LoggingFilter() ); 15 connector.getFilterChain().addLast( "codec", new ProtocolCodecFilter( new TextLineCodecFactory( Charset.forName( "UTF-8" )))); //设置编码过滤器 16 connector.setConnectTimeout(30); 17 connector.setHandler(new TimeClientHandler());//设置事件处理器 18 ConnectFuture cf = connector.connect( 19 new InetSocketAddress("127.0.0.1", 9123));//建立连接 20 cf.awaitUninterruptibly();//等待连接创建完成 21 cf.getSession().write("hello");//发送消息 22 cf.getSession().write("quit");//发送消息 23 cf.getSession().getCloseFuture().awaitUninterruptibly();//等待连接断开 24 connector.dispose(); 25 } 26 }
6.总结
通过上述示例可以了解到:使用MINA框架来开发的网络应用程序代码结构更清晰;MINA框架完成了底层的线程管理;MINA内置的编码器可以满足大多数用户的需求,省去了开发人员消息编码解码的工作。具称使用MINA开发服务器程序的性能已经逼近使用 C/C++ 语言开发的网络服务。因此,建议在网络应用程序开发过程中尝试使用MINA框架来提高我们的开发效率和应用程序的执行效率。
二。
客户端通信过程
1.通过SocketConnector同服务器端建立连接
2.链接建立之后I/O的读写交给了I/O Processor线程,I/O Processor是多线程的
3.通过I/O Processor读取的数据经过IoFilterChain里所有配置的IoFilter,IoFilter进行消息的过滤,格式的转换,在这个层面可以制定一些自定义的协议
4.最后IoFilter将数据交给Handler进行业务处理,完成了整个读取的过程
5.写入过程也是类似,只是刚好倒过来,通过IoSession.write写出数据,然后Handler进行写入的业务处理,处理完成后交给IoFilterChain,进行消息过滤和协议的转换,最后通过I/O Processor将数据写出到socket通道
IoFilterChain作为消息过滤链
1.读取的时候是从低级协议到高级协议的过程,一般来说从byte字节逐渐转换成业务对象的过程
2.写入的时候一般是从业务对象到字节byte的过程
IoSession贯穿整个通信过程的始终
整个过程可以用一个图来表现
消息箭头都是有NioProcessor-N线程发起调用,默认情况下也在NioProcessor-N线程中执行
类图
http://mina.apache.org/class-diagrams.html#ClassDiagrams-ProtocolDecoderclassdiagram
Connector
作为连接客户端,SocketConector用来和服务器端建立连接,连接成功,创建IoProcessor Thread(不能超过指定的processorCount),Thread由指定的线程池进行管理,IoProcessor 利用NIO框架对IO进行处理,同时创建IoSession。连接的建立是通过Nio的SocketChannel进行。
NioSocketConnector connector = new NioSocketConnector(processorCount);
ConnectFuture future = connector.connect(new InetSocketAddress(HOSTNAME, PORT));建立一个I/O通道
Acceptor
作为服务器端的连接接受者,SocketAcceptor用来监听端口,同客户端建立连接,连接建立之后的I/O操作全部交给IoProcessor进行处理
IoAcceptor acceptor = new NioSocketAcceptor();
acceptor.bind( new InetSocketAddress(PORT) );
Protocol
利用IoFilter,对消息进行解码和编码,如以下代码通过 MyProtocolEncoder 将java对象转成byte串,通过MyProtocalDecoder 将byte串恢复成java对象
1 connector.getFilterChain().addLast("codec", new ProtocolCodecFilter(new MyProtocalFactory())); 2 ...... 3 public class MyProtocalFactory implements ProtocolCodecFactory { 4 ProtocolEncoderAdapter encoder = new MyProtocolEncoder(); 5 ProtocolDecoder decoder = new MyProtocalDecoder() ; 6 public ProtocolDecoder getDecoder(IoSession session) throws Exception { 7 return decoder; 8 } 9 public ProtocolEncoder getEncoder(IoSession session) throws Exception { 10 return encoder; 11 } 12 } 13 ...... 14 public class MyProtocalDecoder extends ProtocolDecoderAdapter { 15 16 public void decode(IoSession session, IoBuffer in, ProtocolDecoderOutput out) 17 throws Exception { 18 19 int id = in.getInt(); 20 int len = in.getInt(); 21 byte[] dst = new byte[len]; 22 23 in.get(dst); 24 25 String name = new String(dst,"GBK"); 26 27 Item item = new Item(); 28 item.setId(id); 29 item.setName(name); 30 out.write(item); 31 } 32 } 33 ...... 34 public class MyProtocolEncoder extends ProtocolEncoderAdapter { 35 36 public void encode(IoSession session, Object message, 37 ProtocolEncoderOutput out) throws Exception { 38 Item item = (Item)message; 39 int byteLen = 8 + item.getName().getBytes("GBK").length ; 40 IoBuffer buf = IoBuffer.allocate(byteLen); 41 buf.putInt(item.getId()); 42 buf.putInt(item.getName().getBytes("GBK").length); 43 buf.put(item.getName().getBytes("GBK")); 44 buf.flip(); 45 out.write(buf); 46 47 } 48 }
handler
具体处理事件,事件包括:sessionCreated、sessionOpened、sessionClosed、sessionIdle、exceptionCaught、messageReceived、messageSent。
connector.setHandler(new MyHandler());MyHandler继承IoHandlerAdapter类或者实现IoHandler接口.事件最终由IoProcessor线程发动调用。
Processor
I/O处理器、允许多线程读写,开发过程中只需要指定线程数量,Processor通过Nio框架进行I/O的续写操作,Processor包含了Nio的Selector的引用。这点也正是mina的优势,如果直接用Nio编写,则需要自己编写代码来实现类似Processor的功能。正因为I/O Processor是异步处理读写的,所以我们有时候需要识别同一个任务的消息,比如一个任务包括发送消息,接收消息,反馈消息,那么我们需要在制定消息格式的时候,消息头里能包含一个能识别是同一个任务的id。
I/O Porcessor线程数的设置 :如果是SocketConnector,则可以在构造方法中指定,如:new SocketConnector(processorCount, Executors.newCachedThreadPool());如果是SocketAcceptor,也是一样的:SocketAcceptor acceptor = new SocketAcceptor(ProcessorCount, Executors.newCachedThreadPool());
processorCount为最大Porcessor线程数,这个值可以通过性能测试进行调优,默认值是cpu核数量+1(Runtime.getRuntime().availableProcessors() + 1)。
比较奇怪的是,每个IoProcessor在创建的时候会本地自己和自己建立一个连接?
IoSession
IoSession是用来保持IoService的上下文,一个IoService在建立Connect之后建立一个IoSession(一个连接一个session),IoSession的生命周期从Connection建立到断开为止
IoSession做两件事情:
1.通过IoSession可以获取IoService的所有相关配置对象(持有对IoService,Processor池,SocketChannel,SessionConfig和IoService.IoHandler的引用)
2.通过IoSession.write 是数据写出的入口
关于线程
http://mina.apache.org/configuring-thread-model.html
ThreadModel 1.x版本的mina还有线程模式选项在2.x之后就没有了
1.x版本指定线程模式
SocketConnectorConfig cfg = new SocketConnectorConfig();
cfg.setThreadModel(ThreadModel.MANUAL);
MINA有3种worker线程
Acceptor、Connector、I/O processor 线程
Acceptor Thread
一般作为服务器端链接的接收线程,实现了接口IoService,线程的数量就是创建SocketAcceptor 的数量
Connector Thread
一般作为客户端的请求建立链接线程,实现了接口IoService,维持了一个和服务器端Acceptor的一个链接,线程数量就是创建SocketConnector 的数量
Mina的SocketAcceptor和SocketConnector均是继承了BaseIoService,是对IoService的两种不同的实现
I/O processor Thread
作为I/O真正处理的线程,存在于服务器端和客户端,用来处理I/O的读写操作,线程的数量是可以配置的,默认最大数量是CPU个数+1
服务器端:在创建SocketAcceptor的时候指定ProcessorCount
SocketAcceptor acceptor = new SocketAcceptor(Runtime.getRuntime().availableProcessors() + 1, Executors.newCachedThreadPool());
客户端:在创建SocketConnector 的时候指定ProcessorCount
SocketConnector connector = new SocketConnector(Runtime.getRuntime().availableProcessors() + 1, Executors.newCachedThreadPool());
I/O Processor Thread,是依附于IoService,类似上面的例子SocketConnector connector = new SocketConnector(Runtime.getRuntime().availableProcessors() + 1, Executors.newCachedThreadPool());是指SocketConnector这个线程允许CPU+1个I/O Processor Thread
NioProcessor虽然是多线程,但是对与一个连接的时候业务处理只会使用一个线程进行处理(Processor线程对于一个客户端连接只使用一个线程NioProcessor-n)如果handler的业务比较耗时,会导致NioProcessor线程堵塞 ,在2个客户端同时连接上来的时候会创建第2个(前提是第1个NioProcessor正在忙),创建的最大数量由Acceptor构造方法的时候指定。如果:一个客户端连接同服务器端有很多通信,并且I/O的开销不大,但是Handler处理的业务时间比较长,那么需要采用独立的线程模式,在FilterChain的最后增加一个ExecutorFitler:
acceptor.getFilterChain().addLast("threadPool", new ExecutorFilter(Executors.newCachedThreadPool()));
这样可以保证processor和handler的线程是分开的,否则:客户端发送3个消息,而服务器对于每个消息要处理10s左右,那么这3个消息是被串行处理,在处理第一个消息的时候,后面的消息将被堵塞,同样反过来客户端也有同样的问题。
客户端Porcessor堵塞测试情况:
1.以下代码在建立连接后连续发送了5个消息(item)
1 ConnectFuture future = connector.connect(new InetSocketAddress(HOSTNAME, PORT)); 2 future.awaitUninterruptibly(); 3 session = future.getSession(); 4 Item item = new Item(); 5 item.setId(12345); 6 item.setName("hi"); 7 session.write(item); 8 session.write(item); 9 session.write(item); 10 session.write(item); 11 session.write(item);
2.在handle的messageSent方法进行了延时处理,延时3秒
1 public void messageSent(IoSession session, Object message) throws Exception { 2 Thread.sleep(3000); 3 System.out.println(message); 4 5 }
3.测试结果
5个消息是串行发送,都由同一个IoPorcessor线程处理
1 session.write(item); 2 session.write(item); 3 session.write(item); 4 session.write(item); 5 session.write(item);
4.增加ExecutorFilter,ExecutorFilter保证在处理handler的时候是独立线程
connector.getFilterChain().addLast("threadPool", new ExecutorFilter(Executors.newCachedThreadPool()));
5.测试结果
4个session.wirte变成了并行处理,服务器端同时收到了5条消息