天天看点

socket, nio socket 及nio socket框架MINA总结 (转)

http://blog.csdn.net/lcllcl987/archive/2007/04/16/1566114.aspx

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));

   String str = wt.readLine();

   out.println(str);

   out.flush();

   if (str.equals("end")) {

    break;

   System.out.println(in.readLine());

  server.close();

2.nio:非阻塞通讯模式

2.1NIO 设计背后的基石:反应器模式,用于事件多路分离和分派的体系结构模式。 

反应器模式的核心功能如下:

将事件多路分用 

将事件分派到各自相应的事件处理程序

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");

     socket = (SocketChannel)key.channel();

     if (result.length() > 0 ) {

    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);

            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)

  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);

   nb.startServer();

   System.out.println("the nonBlocking server is started!");

  catch (IOException e)

   System.exit(-1);

 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;

 public void makeConnection() {

  System.getProperties().put("socksProxySet", "true");

  System.getProperties().put("socksProxyHost", proxyHost);

  System.getProperties().put("socksProxyPort", proxyPort);

  int result = 0;

   client = SocketChannel.open();

   isa = new InetSocketAddress(host, port);

   client.connect(isa);

   client.configureBlocking(false);

   receiveMessage();

  } catch (UnknownHostException e) {

  } catch (IOException e) {

  long begin = System.currentTimeMillis();

  sendMessage();

  long end = System.currentTimeMillis();

  long userTime = end - begin;

  System.out.println("use tiem: " + userTime);

   interruptThread();

   System.exit(0);

 public int sendMessage() {

    System.out.println("Inside SendMessage");

  String msg = null;

  ByteBuffer bytebuf;

   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");

   try {

    Thread.sleep(5000);

   } catch (Exception e) {

    e.printStackTrace();

   return -1;

  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);

    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);

   } catch (IOException e) {

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

        acceptor.bind(

                new InetSocketAddress( SERVER_PORT ),

                new ServerSessionHandler( ), cfg ); // 绑定address和handler

client:

        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) {

    public int getStatus() {

        return status;

    public void setStatus(int status) {

    public int getType() {

        return type;

    public void setType(int 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(

                    new ProtocolCodecFilter( new ObjectSerializationCodecFactory() ) );

        IoSession session;

        Message msg = new Message(0,1,"hello");

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;

        IoAcceptor acceptor = new SocketAcceptor();

        // Prepare the service configuration.

        cfg.setReuseAddress( true );

                new ServerSessionHandler( ), cfg );

        System.out.println( "The server Listening on port " + SERVER_PORT );

ServerSessionHandler.java

public class ServerSessionHandler extends IoHandlerAdapter

        // set idle time to 60 seconds

        session.setIdleTime( IdleStatus.BOTH_IDLE, 60 );

        session.setAttribute("times",new Integer(0));

        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

        // close the connection on exceptional situation

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:

http://www.mail-archive.com/[email protected]/msg02252.html

原来需要下载:backport-util-concurrent.jar并加入classpath

http://dcl.mathcs.emory.edu/util/backport-util-concurrent/

继续阅读