nio学习： 最近花了点时间研究了一下nio，及其开源框架MINA，现把心得总结如下： 1：传统socket:阻塞式通信 每建立一个Socket连接时，同时创建一个新线程对该Socket进行单独通信（采用阻塞的方式通信）。 这种方式具有很高的响应速度，并且控制起来也很简单，在连接数较少的时候非常有效，但是如果 对每一个连接都产生一个线程的无疑是对系统资源的一种浪费，如果连接数较多将会出现资源不足的情况 example: server code:
public class MultiUserServer extends Thread {
private Socket client; public MultiUserServer(Socket c) { this.client = c; }public void run() {
try { BufferedReader in = new BufferedReader(new InputStreamReader(client .getInputStream())); PrintWriter out = new PrintWriter(client.getOutputStream()); // Mutil User but can't parallel while (true) { String str = in.readLine(); System.out.println(str); SocketLog.debug("receive message: " + str); out.println("has receive...."); out.flush(); if (str.equals("end")) break; } client.close(); } catch (IOException ex) { } }public static void main(String[] args) throws IOException {
int port = 5678; if (args.length > 0) port = Integer.parseInt(args[0]); ServerSocket server = new ServerSocket(port); SocketLog.debug("the server socket application is created!"); while (true) { // transfer location change Single User or Multi User MultiUserServer mu = new MultiUserServer(server.accept()); mu.start(); } } }client code:
public class Client {
static Socket server;
public static void main(String[] args) throws Exception {
//set socket proxy. String proxyHost = "192.168.254.212"; String proxyPort = "1080"; System.getProperties().put("socksProxySet","true"); System.getProperties().put("socksProxyHost",proxyHost); System.getProperties().put("socksProxyPort",proxyPort); String host = "132.201.69.80"; int port = 13086; if (args.length > 1) { host = args[0]; port = Integer.parseInt(args[1]); } System.out.println("connetioning:" + host + ":" + port); server = new Socket(host, port); BufferedReader in = new BufferedReader(new InputStreamReader(server .getInputStream())); PrintWriter out = new PrintWriter(server.getOutputStream()); BufferedReader wt = new BufferedReader(new InputStreamReader(System.in)); while (true) { String str = wt.readLine(); out.println(str); out.flush(); if (str.equals("end")) { break; } System.out.println(in.readLine()); } server.close(); } }2.nio:非阻塞通讯模式
２．１NIO 设计背后的基石：反应器模式，用于事件多路分离和分派的体系结构模式。 反应器模式的核心功能如下： 将事件多路分用 将事件分派到各自相应的事件处理程序NIO 的非阻塞 I/O 机制是围绕 选择器和 通道构建的。 Channel 类表示服务器和客户机之间的
一种通信机制。Selector 类是 Channel 的多路复用器。 Selector 类将传入客户机请求多路分 用并将它们分派到各自的请求处理程序。 通道(Channel 类)：表示服务器和客户机之间的一种通信机制。 选择器(Selector类)：是 Channel 的多路复用器。Selector 类将传入的客户机请求多路分用并将它们 分派到各自的请求处理程序。简单的来说：
NIO是一个基于事件的IO架构，最基本的思想就是：有事件我通知你，你再去做你的事情．
而且NIO的主线程只有一个，不像传统的模型，需要多个线程以应对客户端请求，也减轻 了JVM的工作量。 当Channel注册至Selector以后，经典的调用方法如下：while (somecondition) {
int n = selector.select(TIMEOUT); if (n == 0) continue; for (Iterator iter = selector.selectedKeys().iterator(); iter .hasNext();) { if (key.isAcceptable()) doAcceptable(key); if (key.isConnectable()) doConnectable(key); if (key.isValid() && key.isReadable()) doReadable(key); if (key.isValid() && key.isWritable()) doWritable(key); iter.remove(); } } nio中取得事件通知，就是在selector的select事件中完成的。在selector事件时有一个线程 向操作系统询问，selector中注册的Channel&&SelectionKey的键值对的各种事件是否有发生， 如果有则添加到selector的selectedKeys属性Set中去，并返回本次有多少个感兴趣的事情发生。 如果发现这个值>0，表示有事件发生，马上迭代selectedKeys中的SelectionKey， 根据Key中的表示的事件，来做相应的处理。 实际上，这段说明表明了异步socket的核心，即异步socket不过是将多个socket的调度（或者还有他们的线程调度） 全部交给操作系统自己去完成，异步的核心Selector，不过是将这些调度收集、分发而已。 2.2 nio example: server code： public class NonBlockingServer { public Selector sel = null; public ServerSocketChannel server = null; public SocketChannel socket = null; public int port = 4900; String result = null; public NonBlockingServer() { System.out.println("Inside default ctor"); } public NonBlockingServer(int port) { System.out.println("Inside the other ctor"); this.port = port; }public void initializeOperations() throws IOException,UnknownHostException
{ System.out.println("Inside initialization"); sel = Selector.open(); server = ServerSocketChannel.open(); server.configureBlocking(false); InetAddress ia = InetAddress.getLocalHost(); InetSocketAddress isa = new InetSocketAddress(ia,port); server.socket().bind(isa); } public void startServer() throws IOException { System.out.println("Inside startserver"); initializeOperations(); System.out.println("Abt to block on select()"); SelectionKey acceptKey = server.register(sel, SelectionKey.OP_ACCEPT ); while (acceptKey.selector().select() > 0 ) { Set readyKeys = sel.selectedKeys(); Iterator it = readyKeys.iterator();while (it.hasNext()) {
SelectionKey key = (SelectionKey)it.next(); it.remove(); if (key.isAcceptable()) { System.out.println("Key is Acceptable"); ServerSocketChannel ssc = (ServerSocketChannel) key.channel(); socket = (SocketChannel) ssc.accept(); socket.configureBlocking(false); SelectionKey another = socket.register(sel,SelectionKey.OP_READ|SelectionKey.OP_WRITE); } if (key.isReadable()) { System.out.println("Key is readable"); String ret = readMessage(key); if (ret.length() > 0) { writeMessage(socket,ret); } } if (key.isWritable()) { System.out.println("THe key is writable"); String ret = readMessage(key); socket = (SocketChannel)key.channel(); if (result.length() > 0 ) { writeMessage(socket,ret); } } } } }public void writeMessage(SocketChannel socket,String ret)
{ System.out.println("Inside the loop");if (ret.equals("quit") || ret.equals("shutdown")) {
return; } try {String s = "This is context from server!-----------------------------------------";
Charset set = Charset.forName("us-ascii"); CharsetDecoder dec = set.newDecoder(); CharBuffer charBuf = dec.decode(ByteBuffer.wrap(s.getBytes())); System.out.println(charBuf.toString()); int nBytes = socket.write(ByteBuffer.wrap((charBuf.toString()).getBytes())); System.out.println("nBytes = "+nBytes); result = null; } catch(Exception e) { e.printStackTrace(); }}
public String readMessage(SelectionKey key) { int nBytes = 0; socket = (SocketChannel)key.channel(); ByteBuffer buf = ByteBuffer.allocate(1024); try { nBytes = socket.read(buf); buf.flip(); Charset charset = Charset.forName("us-ascii"); CharsetDecoder decoder = charset.newDecoder(); CharBuffer charBuffer = decoder.decode(buf); result = charBuffer.toString(); } catch(IOException e) { e.printStackTrace(); } return result; }public static void main(String args[])
{ NonBlockingServer nb; if (args.length < 1) { nb = new NonBlockingServer(); } else { int port = Integer.parseInt(args[0]); nb = new NonBlockingServer(port); } try { nb.startServer(); System.out.println("the nonBlocking server is started!"); } catch (IOException e) { e.printStackTrace(); System.exit(-1); } } }client code:
public class Client {
public SocketChannel client = null;public InetSocketAddress isa = null;
public RecvThread rt = null;
private String host;
private int port;
public Client(String host, int port) {
this.host = host; this.port = port; }public void makeConnection() {
String proxyHost = "192.168.254.212"; String proxyPort = "1080"; System.getProperties().put("socksProxySet", "true"); System.getProperties().put("socksProxyHost", proxyHost); System.getProperties().put("socksProxyPort", proxyPort);int result = 0;
try { client = SocketChannel.open(); isa = new InetSocketAddress(host, port); client.connect(isa); client.configureBlocking(false); receiveMessage(); } catch (UnknownHostException e) { e.printStackTrace(); } catch (IOException e) { e.printStackTrace(); } long begin = System.currentTimeMillis();sendMessage();
long end = System.currentTimeMillis();
long userTime = end - begin; System.out.println("use tiem: " + userTime); try { interruptThread(); client.close(); System.exit(0); } catch (IOException e) { e.printStackTrace(); } }public int sendMessage() {
System.out.println("Inside SendMessage"); String msg = null; ByteBuffer bytebuf; int nBytes = 0; try { msg = "It's message from client!"; System.out.println("msg is "+msg); bytebuf = ByteBuffer.wrap(msg.getBytes()); for (int i = 0; i < 1000; i++) { nBytes = client.write(bytebuf); System.out.println(i + " finished"); } interruptThread(); try { Thread.sleep(5000); } catch (Exception e) { e.printStackTrace(); } client.close(); return -1;} catch (IOException e) {
e.printStackTrace(); }return nBytes;
}
public void receiveMessage() {
rt = new RecvThread("Receive THread", client); rt.start();}
public void interruptThread() {
rt.val = false; }public static void main(String args[]) {
if (args.length < 2) { System.err.println("You should put 2 args: host,port"); } else { String host = args[0]; int port = Integer.parseInt(args[1]); Client cl = new Client(host, port); cl.makeConnection(); } BufferedReader in = new BufferedReader(new InputStreamReader(System.in)); String msg;}
public class RecvThread extends Thread {
public SocketChannel sc = null;public boolean val = true;
public RecvThread(String str, SocketChannel client) {
super(str); sc = client; }public void run() {
int nBytes = 0; ByteBuffer buf = ByteBuffer.allocate(2048); try { while (val) { while ((nBytes = nBytes = client.read(buf)) > 0) { buf.flip(); Charset charset = Charset.forName("us-ascii"); CharsetDecoder decoder = charset.newDecoder(); CharBuffer charBuffer = decoder.decode(buf); String result = charBuffer.toString(); System.out.println("the server return: " + result); buf.flip();}
}} catch (IOException e) {
e.printStackTrace();}
}
} } 3：Socket网络框架 MINA MINA是一个网络应用框架，在不牺牲性能和可扩展性的前提下用于解决如下问题： 1：快速开发自己的英勇。 2：高可维护性，高可复用性：网络I/O编码，消息的编/解码，业务逻辑互相分离。 3：相对容易的进行单元测试。
3.1 IoFilters：
IoFilter为MINA的功能扩展提供了接口。它拦截所有的IO事件进行事件的预处理和后处理(AOP)。我们可以把它想象成 Servlet的filters。 IoFilter能够实现以下几种目的： 事件日志 性能检测 数据转换(e.g. SSL support)，codec 防火墙…等等3.2 codec: ProtocolCodecFactory
MINA提供了方便的Protocol支持。如上说讲，codec在IoFilters中设置。 通过它的Encoder和Decoder，可以方便的扩展并支持各种基于Socket的网络协议，比如HTTP服务器、FTP服务器、Telnet服务器等等。要实现自己的编码/×××(codec)只需要实现interface: ProtocolCodecFactory即可.
在MINA 1.0版本，MINA已经实现了几个常用的(codec factory):DemuxingProtocolCodecFactory,
NettyCodecFactory, ObjectSerializationCodecFactory, TextLineCodecFactory 其中： TextLineCodecFactory: A ProtocolCodecFactory that performs encoding and decoding between a text line data and a Java string object. This codec is useful especially when you work with a text-based protocols such as SMTP and IMAP.ObjectSerializationCodecFactory:
A ProtocolCodecFactory that serializes and deserializes Java objects. This codec is very useful when you have to prototype your application rapidly without any specific codec.DemuxingProtocolCodecFactory：
A composite ProtocolCodecFactory that consists of multiple MessageEncoders and MessageDecoders. ProtocolEncoder and ProtocolDecoder this factory returns demultiplex incoming messages and buffers to appropriate MessageEncoders and MessageDecoders.NettyCodecFactory:
A MINA ProtocolCodecFactory that provides encoder and decoder for Netty2 Messages and MessageRecognizers.3.3 business logic: IoHandler
MINA中，所有的业务逻辑都有实现了IoHandler的class完成
interfaceHandles: all protocol events fired by MINA. There are 6 event handler methods, and they are all invoked by MINA automatically. 当事件发生时，将触发IoHandler中的方法: sessionCreated, sessionOpened, sessionClosed, sessionIdle, exceptionCaught, messageReceived, messageSent MINA 1.O中，IoHandler的实现类： ChainedIoHandler, DemuxingIoHandler, IoHandlerAdapter, SingleSessionIoHandlerDelegate, StreamIoHandler 具体细节可参考javadoc。3.4 MINA的高级主题：线程模式
MINA通过它灵活的filter机制来提供多种线程模型。 没有线程池过滤器被使用时MINA运行在一个单线程模式。 如果添加了一个IoThreadPoolFilter到IoAcceptor，将得到一个leader-follower模式的线程池。 如果再添加一个ProtocolThreadPoolFilter，server将有两个线程池； 一个(IoThreadPoolFilter)被用于对message对象进行转换，另外一个(ProtocolThreadPoolFilter)被用于处理业务逻辑。 SimpleServiceRegistry加上IoThreadPoolFilter和ProtocolThreadPoolFilter的缺省实现即可适用于需 要高伸缩性的应用。如果想使用自己的线程模型，请参考SimpleServiceRegistry的源代码，并且自己初始化Acceptor。
IoThreadPoolFilter threadPool = new IoThreadPoolFilter();threadPool.start();
IoAcceptor acceptor = new SocketAcceptor(); acceptor.getFilterChain().addLast( "threadPool", threadPool); ProtocolThreadPoolFilter threadPool2 = new ProtocolThreadPoolFilter(); threadPool2.start(); ProtocolAcceptor acceptor2 = new IoProtocolAcceptor( acceptor ); acceptor2.getFilterChain().addLast( "threadPool", threadPool2 ); ... threadPool2.stop(); threadPool.stop(); 采用MINA进行socket开发，一般步骤如下： 1： server: IoAcceptor acceptor = new SocketAcceptor(); //建立client接收器 or client: SocketConnector connector = new SocketConnector(); //建立一个连接器 2：server的属性配置： SocketAcceptorConfig cfg = new SocketAcceptorConfig(); cfg.setReuseAddress(true); cfg.getFilterChain().addLast( "codec", new ProtocolCodecFilter( new ObjectSerializationCodecFactory() ) ); //对象序列化 codec factory cfg.getFilterChain().addLast( "logger", new LoggingFilter() ); 3：绑定address和business logic server: acceptor.bind( new InetSocketAddress( SERVER_PORT ), new ServerSessionHandler( ), cfg ); // 绑定address和handlerclient:
connector.connect(new InetSocketAddress( HOSTNAME, PORT ), new ClientSessionHandler(msg), cfg );下面的这个简单的example演示client和server传递object的过程：
Message.java public class Message implements Serializable {private int type;
private int status; private String msgBody; public Message(int type, int status, String msgBody) { this.type = type; this.status = status; this.msgBody = msgBody; }public String getMsgBody() {
return msgBody; }public void setMsgBody(String msgBody) {
this.msgBody = msgBody; }public int getStatus() {
return status; }public void setStatus(int status) {
this.status = status; }public int getType() {
return type; }public void setType(int type) {
this.type = type; } }Client.java
public class Client { private static final String HOSTNAME = "localhost"; private static final int PORT = 8080; private static final int CONNECT_TIMEOUT = 30; // seconds public static void main( String[] args ) throws Throwable { SocketConnector connector = new SocketConnector(); // Configure the service. SocketConnectorConfig cfg = new SocketConnectorConfig(); cfg.setConnectTimeout( CONNECT_TIMEOUT ); cfg.getFilterChain().addLast( "codec", new ProtocolCodecFilter( new ObjectSerializationCodecFactory() ) );cfg.getFilterChain().addLast( "logger", new LoggingFilter() );
IoSession session; Message msg = new Message(0,1,"hello"); connector.connect(new InetSocketAddress( HOSTNAME, PORT ), new ClientSessionHandler(msg), cfg );}
}ClientSessionHandler.java
public class ClientSessionHandler extends IoHandlerAdapter { private Object msg; public ClientSessionHandler(Object msg) { this.msg = msg; } public void sessionOpened( IoSession session ) { session.write(this.msg); }public void messageReceived( IoSession session, Object message )
{ System.out.println("in messageReceived!"); Message rm = (Message ) message; SessionLog.debug(session, rm.getMsgBody()); System.out.println("message is: " + rm.getMsgBody()); session.write(rm); }public void exceptionCaught( IoSession session, Throwable cause )
{ session.close(); } }Server.java
public class Server { private static final int SERVER_PORT = 8080;public static void main( String[] args ) throws Throwable
{ IoAcceptor acceptor = new SocketAcceptor(); // Prepare the service configuration. SocketAcceptorConfig cfg = new SocketAcceptorConfig(); cfg.setReuseAddress( true );cfg.getFilterChain().addLast(
"codec", new ProtocolCodecFilter( new ObjectSerializationCodecFactory() ) ); cfg.getFilterChain().addLast( "logger", new LoggingFilter() );acceptor.bind(
new InetSocketAddress( SERVER_PORT ), new ServerSessionHandler( ), cfg );System.out.println( "The server Listening on port " + SERVER_PORT );
} }ServerSessionHandler.java
public class ServerSessionHandler extends IoHandlerAdapter { public void sessionOpened( IoSession session ) { // set idle time to 60 seconds session.setIdleTime( IdleStatus.BOTH_IDLE, 60 ); session.setAttribute("times",new Integer(0)); }public void messageReceived( IoSession session, Object message )
{ System.out.println("in messageReceived"); int times = ((Integer)(session.getAttribute("times"))).intValue(); System.out.println("tiems = " + times); // communicate 30 times,then close the session. if (times < 30) { times++; session.setAttribute("times", new Integer(times)); Message msg; msg = (Message) message; msg.setMsgBody("in server side: " + msg.getMsgBody()); System.out.println("begin send msg: " + msg.getMsgBody()); session.write(msg); } else { session.close(); } }public void sessionIdle( IoSession session, IdleStatus status )
{ SessionLog.info( session, "Disconnecting the idle." ); // disconnect an idle client session.close(); }public void exceptionCaught( IoSession session, Throwable cause )
{ // close the connection on exceptional situation session.close(); } } MINA自己附带的Demo已经很好的说明了它的运用。 值得一提的是它的SumUp：客户端发送几个数字，服务端求和后并返回结果。这个简单的程序演示了如何自己实现CODEC。补充提示：
下载并运行MINA的demo程序还颇非周折: 运行MINA demo appli擦tion： 1:在JDK5 产生错误: Exception in thread "main" java.lang.NoClassDefFoundError: edu/emory/mathcs/backport/java/util/concurrent/Executor at org.apache.mina.example.reverser.Main.main(Main.java:44) 察看mina的QA email:原来需要下载:backport-util-concurrent.jar并加入classpath
继续运行还是报错:
Exception in thread "main" java.lang.NoClassDefFoundError: org/slf4j/LoggerFactory原来MINA采用了slf4j项目作为log，继续下载
slf4j-simple.jar等，并加入classpath:posted on 2007-06-06 14:31 阅读(5923) 所属分类: