Mina框架学习笔记（二）

[size=medium]上一篇只写了一个服务端。这一次来构建一个客户端。



首先，在引入 上一篇中讲到的几个jar包处，还要引入一个mina-example-2.0.0.jar



本程序的主要功能是，客户端向服务器发送几个数字，然后服务器给客户端 返回结果。文字就不多写了，我在上面都写了注释！



下面的服务端代码：

import java.io.IOException;  
import java.net.InetSocketAddress;  
import org.apache.mina.example.sumup.codec.SumUpProtocolCodecFactory;  
import org.apache.mina.filter.codec.ProtocolCodecFilter;  
import org.apache.mina.filter.codec.serialization.ObjectSerializationCodecFactory;  
import org.apache.mina.filter.logging.LoggingFilter;  
import org.apache.mina.transport.socket.nio.NioSocketAcceptor;  
public class MinaServer {  
    private static final int PORT = 8389,BUF_SIZE = 2048;  
    // Set this to false to use object serialization instead of custom codec.  
    private static final boolean USE_CUSTOM_CODEC = true;  
    public static void main(String[] args) throws IOException {  
        NioSocketAcceptor acceptor = new NioSocketAcceptor();  
          
        // Add 'codec' filter  
        if(USE_CUSTOM_CODEC){  
            acceptor.getFilterChain().addLast("codec",  
                    new ProtocolCodecFilter(  
                            new SumUpProtocolCodecFactory(false)));  
        } else {  
            acceptor.getFilterChain().addLast("codec",  
                    new ProtocolCodecFilter(  
                            new ObjectSerializationCodecFactory()));  
        }  
          
        //This filter will log all information such as newly created   
            //sessions, messages received, messages sent, session closed  
        acceptor.getFilterChain().addLast("logger", new LoggingFilter());  
          
        acceptor.setHandler(new MinaServerHandler());  
          
        acceptor.getSessionConfig().setReadBufferSize(BUF_SIZE);  
        //the first parameter defines what actions to check for when   
            //determining if a session is idle, the second parameter defines   
            //the length of time in seconds that must occur before a session   
            //is deemed to be idle.  
        //acceptor.getSessionConfig().setIdleTime(IdleStatus.BOTH_IDLE, 60);  
          
        acceptor.bind(new InetSocketAddress(PORT));  
        System.out.println("Listening on port:"+PORT);  
          
    }  
}  

它使用到了一个自定义的Handler来管理连接过程中的各个事件，代码如下：

import org.apache.mina.core.service.IoHandlerAdapter;  
import org.apache.mina.core.session.IdleStatus;  
import org.apache.mina.core.session.IoSession;  
import org.apache.mina.example.sumup.ServerSessionHandler;  
import org.apache.mina.example.sumup.message.AddMessage;  
import org.apache.mina.example.sumup.message.ResultMessage;  
import org.slf4j.Logger;  
import org.slf4j.LoggerFactory;  
public class MinaServerHandler extends IoHandlerAdapter {  
    private static final String SUM_KEY = "sum";  
    private final static Logger LOGGER = LoggerFactory.getLogger(ServerSessionHandler.class);  
      
    @Override  
    public void exceptionCaught(IoSession session, Throwable cause)  
            throws Exception {  
        session.close(true);  
        cause.printStackTrace();  
    }  
    @Override  
    public void messageReceived(IoSession session, Object message)  
            throws Exception {  
        // client only sends AddMessage. otherwise, we will have to identify  
        // its type using instanceof operator.  
        AddMessage am = (AddMessage) message;  
          
        //Add the value to the current sum.  
        int sum = ((Integer) session.getAttribute(SUM_KEY)).intValue();  
        int value = am.getValue();  
        long expectedSum = (long) sum+value;  
          
        if(expectedSum > Integer.MAX_VALUE || expectedSum < Integer.MIN_VALUE){  
            //If the sum overflows or underflows , return error message.  
            ResultMessage rMessage = new ResultMessage();  
            rMessage.setSequence(am.getSequence()); // copy sequence  
            rMessage.setOk(false);  
            session.write(rMessage);  
        } else {  
            //sum up  
            sum = (int) expectedSum;  
            session.setAttribute(SUM_KEY,new Integer(sum));  
              
            //return the result message.  
            ResultMessage rmMessage = new ResultMessage();  
            rmMessage.setSequence(am.getSequence()); //copy sequece  
              
            rmMessage.setOk(true);  
            rmMessage.setValue(sum);  
            session.write(rmMessage);  
        }  
    }  
    @Override  
    public void messageSent(IoSession session, Object message) throws Exception {  
        System.out.println("Message sent:"+message);  
    }  
    @Override  
    public void sessionIdle(IoSession session, IdleStatus status)  
            throws Exception {  
        LOGGER.info("Disconnecting the idle...");  
        //disconnect an idle client  
        session.close(true);  
    }  
    @Override  
    public void sessionOpened(IoSession session) throws Exception {  
        //Set idle time to 60 seconds  
        session.getConfig().setIdleTime(IdleStatus.BOTH_IDLE, 60);  
          
        //Initial sum is 0  
        session.setAttribute(SUM_KEY,new Integer(0));  
    }  
}  

接下去就是客户端了。一样的，先写一个主类，然后再写一上Handler来管理事件。

import java.net.InetSocketAddress;  
import org.apache.mina.core.RuntimeIoException;  
import org.apache.mina.core.future.ConnectFuture;  
import org.apache.mina.core.session.IoSession;  
import org.apache.mina.example.sumup.codec.SumUpProtocolCodecFactory;  
import org.apache.mina.filter.codec.ProtocolCodecFilter;  
import org.apache.mina.filter.codec.serialization.ObjectSerializationCodecFactory;  
import org.apache.mina.filter.logging.LoggingFilter;  
import org.apache.mina.transport.socket.nio.NioSocketConnector;  
public class MinaClient {  
    private final static long DEF_TIMEOUT = 60*1000L; //1 minute      
    // Set this to false to use object serialization instead of custom codec.  
    private static final boolean USE_CUSTOM_CODEC = true;  
    // Server and port  
    private static final int PORT = 8389;  
    private static final String SERVER = "127.0.0.1";  
    private static IoSession session;  
      
    /** 
     * @param args 
     * @throws InterruptedException  
     */  
    public static void main(String[] args) throws InterruptedException {  
        if (args.length == 0) {  
           System.out.println("Please specify the list of any integers");  
           return;  
        }  
          
        //prepare values to sum up  
        int len = args.length;  
        int[] values = new int[len];  
        for(int i = 0; i < len; i ++){  
            values[i] = Integer.parseInt(args[i]);   
        }  
          
        // Create TCP/IP connector.  
        NioSocketConnector connector = new NioSocketConnector();  
          
        //Set connect timeout  
        connector.setConnectTimeoutMillis(DEF_TIMEOUT);  
          
        // Add 'codec' filter  
        if(USE_CUSTOM_CODEC){  
            connector.getFilterChain().addLast("codec",  
                    new ProtocolCodecFilter(  
                            new SumUpProtocolCodecFactory(false)));  
        } else {  
            connector.getFilterChain().addLast("codec",  
                    new ProtocolCodecFilter(  
                            new ObjectSerializationCodecFactory()));  
        }  
          
        connector.getFilterChain().addLast("logger", new LoggingFilter());  
        //Start communication  
        connector.setHandler(new NetCatProtocolHandler(values));  
          
        //If it fails to connect to the server,  
        //retry it after 10 seconds!  
        while(true){  
            try{  
                ConnectFuture future = connector.connect(new InetSocketAddress(SERVER,PORT));  
                future.awaitUninterruptibly();  
                session = future.getSession();  
                break;  
            } catch (RuntimeIoException e) {  
                System.err.println("Fail to connect!");  
                e.printStackTrace();  
                Thread.sleep(10*1000L);  
            }  
        }  
          
        //Wait for the connection attempt to be finished.  
        session.getCloseFuture().awaitUninterruptibly();  
          
        connector.dispose();  
    }  
}  

Handler类

import org.apache.mina.core.service.IoHandlerAdapter;  
import org.apache.mina.core.session.IdleStatus;  
import org.apache.mina.core.session.IoSession;  
import org.apache.mina.example.sumup.ClientSessionHandler;  
import org.apache.mina.example.sumup.message.AddMessage;  
import org.apache.mina.example.sumup.message.ResultMessage;  
import org.slf4j.LoggerFactory;  
public class NetCatProtocolHandler extends IoHandlerAdapter {  
      
    private int[] values;     
    private boolean finished;  
    private static final org.slf4j.Logger LOGGER = LoggerFactory.getLogger(ClientSessionHandler.class);  
      
    // provide a method for other class to judge whether it's finished.  
    public boolean isFinished() {  
        return finished;  
    }  
    public NetCatProtocolHandler(int[] values) {  
        this.values = values;  
    }  
    @Override  
    public void exceptionCaught(IoSession session, Throwable cause)  
            throws Exception {  
        session.close(true);  
    }  
    @Override  
    public void messageReceived(IoSession session, Object message)  
            throws Exception {  
//      IoBuffer buffer = (IoBuffer) message;         
//      while(buffer.hasRemaining()){  
//          System.out.println(buffer.getChar());  
//      }         
//      System.out.flush();  
          
          
        // server only sends ResultMessage. otherwise, we will have to identify  
        // its type using instanceof operator.  
        ResultMessage rm = (ResultMessage)message;  
        if(rm.isOk()){  // server returned OK code.           
            // if received the result message which has the last sequence  
            // number, it is time to disconnect.  
            if(rm.getSequence() == values.length - 1){  
                //Print the sum and disconnect.  
                LOGGER.warn("Server error, disconnecting...");  
                session.close(true);  
                  
                finished = true;  
            }  
        }  
    }  
    @Override  
    public void messageSent(IoSession session, Object message) throws Exception {  
        session.write(message);  
        System.out.println("Message sent:"+message);  
    }  
    @Override  
    public void sessionClosed(IoSession session) throws Exception {  
        System.err.println("Total "+ session.getReadBytes()+" byte(s)");  
    }  
    @Override  
    public void sessionIdle(IoSession session, IdleStatus status)  
            throws Exception {  
        if(status == IdleStatus.READER_IDLE) {  
            session.close(true);  
        }  
    }  
    @Override  
    public void sessionOpened(IoSession session) throws Exception {  
        // Set reader idle time to 60 seconds.  
        // sessionIdle(...) method will be invoked when no data is read  
        // for 60 seconds.  
        //session.getConfig().setIdleTime(IdleStatus.READER_IDLE, 60);  
          
        // send summation requests  
        for(int i = 0; i < values.length; i ++){  
            AddMessage message = new AddMessage();  
            message.setSequence(i);  
            message.setValue(values[i]);  
              
            session.write(message);  
        }  
    }  
      
}  

[/size]