mina 传输java对象

mina client和server端传输和接收java对象，是java序列化和反序列化的过程。
mina-core包中有对这一块encoder-decoder的编解码类。
ObjectSerializationCodecFactory是工厂类，在client和server端配置coder filter时使用到。ObjectSerializationEncoder是序列化编码类，ObjectSerializationDecoder是序列化解码类

先看运行代码

server端

		IoAcceptor accepter = new NioSocketAcceptor();
		ProtocolCodecFilter coderFilter = 
	       //ObjectSerializationCodecF
		new ProtocolCodecFilter(new ObjectSerializationCodecFactory());
		accepter.getFilterChain().addLast("a", new LoggingFilter());
		accepter.getFilterChain().addLast("b",coderFilter);
		
		accepter.setHandler(new ObjectServerHandler());
		
		accepter.getSessionConfig().setReadBufferSize(2048);
		accepter.getSessionConfig().setIdleTime(IdleStatus.BOTH_IDLE, 10);
		accepter.bind(new InetSocketAddress(8484));

传输的简单java对象：

public class SimpleObj implements Serializable {

	/****/
	private static final long serialVersionUID = 8217287396365894807L;
	
	private String name;
	private String count;
	public String getName() {
		return name;
	}
	public void setName(String name) {
		this.name = name;
	}
	public String getCount() {
		return count;
	}
	public void setCount(String count) {
		this.count = count;
	}
}

server端handler

public class ObjectServerHandler extends IoHandlerAdapter{
	
    @Override
    public void exceptionCaught( IoSession session, Throwable cause ) throws Exception
    {
        cause.printStackTrace();
        session.close(true);
    }
    @Override
    public void messageReceived( IoSession session, Object message ) throws Exception
    {
    	SimpleObj obj = (SimpleObj) message;
    	System.out.println(obj.getName()+":"+obj.getCount());
        session.write(obj);
    }
}

client端handler

public class ObjectClientHandler extends IoHandlerAdapter {
	private SimpleObj obj;
	
	public ObjectClientHandler(SimpleObj obj){
		this.obj = obj;
	}
	@Override
	public void sessionOpened(IoSession session) throws Exception {
		session.write(obj);
	}
	
	@Override
	public void messageReceived(IoSession session, Object message)
			throws Exception {
		System.out.println(message.toString());
	}
	
	@Override
	public void exceptionCaught(IoSession session, Throwable cause)
			throws Exception {
		session.close(true);
	}
}

client端

		NioSocketConnector connector = new NioSocketConnector();
		connector.setConnectTimeoutMillis(20000);
		connector.getFilterChain().addLast("codes", new ProtocolCodecFilter(
				new ObjectSerializationCodecFactory()));
		SimpleObj obj = new SimpleObj();
		obj.setName("bird");
		obj.setCount("7");
		connector.setHandler(new ObjectClientHandler(obj));
		
		IoSession session = null;
		ConnectFuture future = connector.connect(new InetSocketAddress("localhost", 8484));
		future.awaitUninterruptibly();
		session = future.getSession();
		session.getCloseFuture().awaitUninterruptibly();
		connector.dispose();

先启动server端服务，运行client端，server端就可收到传输的object。打印出bird：7


ObjectSerializationEncoder的encoder方法是序列化开始
org.apache.mina.filter.codec.serialization.ObjectSerializationEncoder

    public void encode(IoSession session, Object message, ProtocolEncoderOutput out) throws Exception {
     //没序列化，直接抛异常
        if (!(message instanceof Serializable)) {
            throw new NotSerializableException();
        }

        IoBuffer buf = IoBuffer.allocate(64);
        buf.setAutoExpand(true);
	//A 将Object序列化后字节内容填充到buffer内
        buf.putObject(message);
	//对象序列化后字节长度，大于最大值就抛异常，一般应该没这么大
	//maxObjectSize = Integer.MAX_VALUE; 
        int objectSize = buf.position() - 4;
        if (objectSize > maxObjectSize) {
            throw new IllegalArgumentException("The encoded object is too big: " + objectSize + " (> " + maxObjectSize
                    + ')');
        }

        buf.flip();
        out.write(buf);
    }

A处方法源码 org.apache.mina.core.buffer.AbstractIoBuffer

   public IoBuffer putObject(Object o) {
        int oldPos = position();
        skip(4); // 前4个长度先跳过，预留存储对象长度
        try {
            ObjectOutputStream out = new ObjectOutputStream(asOutputStream()//B,下面介绍) {
	    //重写了ObjectOutputStream的一个方法，不知道干什么用，
                @Override
                protected void writeClassDescriptor(ObjectStreamClass desc) throws IOException {
                    try {
                        Class<?> clz = Class.forName(desc.getName());
                        if (!Serializable.class.isAssignableFrom(clz)) { // NON-Serializable class
                            write(0);
                            super.writeClassDescriptor(desc);
                        } else { // Serializable class
                            write(1);
                            writeUTF(desc.getName());
                        }
                    } catch (ClassNotFoundException ex) { // Primitive types
                        write(0);
                        super.writeClassDescriptor(desc);
                    }
                }
            };
	    //和普通的java序列化一样，调用writeObject方法，java序列化前面文章有说过
            out.writeObject(o);
            out.flush();
        } catch (IOException e) {
            throw new BufferDataException(e);
        }

        // 填写前面预留的长度位
        int newPos = position();//当前position
        position(oldPos);//position标到开始位0
        putInt(newPos - oldPos - 4);//将object字节长度写入
        position(newPos);//postion到buffer的结尾
        return this;
    }

B处 asOutputStream()方法，就是实例化一个OutputStream,
两个写方法都是把内容放到IoBuffer里。
后续的Stream包装类，也就是ObjectOutputStream不管是writeShort,writeUTF等等也好，
最后都是调用这两个方法，所以上面的writeObject最后都是把object的内容写到了
IoBuffer里

    public OutputStream asOutputStream() {
        return new OutputStream() {
            @Override
            public void write(byte[] b, int off, int len) {
                AbstractIoBuffer.this.put(b, off, len);
            }

            @Override
            public void write(int b) {
                AbstractIoBuffer.this.put((byte) b);
            }
        };
    }

所以encode完后，传到server端的buffer内容
buffer=4个长度位+object序列化后的字节内容（序列化稍微差别，上面重写了个方法）

C处out方法源码
org.apache.mina.filter.codec.AbstractProtocolEncoderOutput

 

  public void write(Object encodedMessage) {
        if (encodedMessage instanceof IoBuffer) {
            IoBuffer buf = (IoBuffer) encodedMessage;
            if (buf.hasRemaining()) {
                messageQueue.offer(buf);
            } else {
                throw new IllegalArgumentException("buf is empty. Forgot to call flip()?");
            }
        } else {
            messageQueue.offer(encodedMessage);
            buffersOnly = false;
        }
    }

下面看server端解码：
org.apache.mina.filter.codec.serialization.ObjectSerializationDecoder

    protected boolean doDecode(IoSession session, IoBuffer in, ProtocolDecoderOutput out) throws Exception {
	//不足4个长度就直接不处理了，前面encode的过程肯定>4
        if (!in.prefixedDataAvailable(4, maxObjectSize)) {
            return false;
        }
	//AA 首先从buffer里吧序列化的对象读出来，反序列化
	//依次读取出来
        out.write(in.getObject(classLoader));
        return true;
    }

AA处getObject方法源码

    public Object getObject(final ClassLoader classLoader) throws ClassNotFoundException {
        if (!prefixedDataAvailable(4)) {
            throw new BufferUnderflowException();
        }
	//读取一个int，也就是读取前面encode的标示位
        int length = getInt();
        if (length <= 4) {
            throw new BufferDataException("Object length should be greater than 4: " + length);
        }

        int oldLimit = limit();
        limit(position() + length);
        try {
            ObjectInputStream in = new ObjectInputStream(asInputStream()//BB) {
		//略去两个重写ObjectInputStream方法
            };
	    //和普通java反序列化一样调用ObjectInputStream.readObject()方法
            return in.readObject();
        } catch (IOException e) {
            throw new BufferDataException(e);
        } finally {
            limit(oldLimit);
        }
    }

  BB处源码org.apache.mina.core.buffer.AbstractIoBuffer
    也是从buffer中读取序列化内容

       public InputStream asInputStream() {
        return new InputStream() {
            @Override
            public int available() {
                return AbstractIoBuffer.this.remaining();
            }

            @Override
            public synchronized void mark(int readlimit) {
                AbstractIoBuffer.this.mark();
            }

            @Override
            public boolean markSupported() {
                return true;
            }

            @Override
            public int read() {
                if (AbstractIoBuffer.this.hasRemaining()) {
                    return AbstractIoBuffer.this.get() & 0xff;
                }

                return -1;
            }

            @Override
            public int read(byte[] b, int off, int len) {
                int remaining = AbstractIoBuffer.this.remaining();
                if (remaining > 0) {
                    int readBytes = Math.min(remaining, len);
                    AbstractIoBuffer.this.get(b, off, readBytes);
                    return readBytes;
                }

                return -1;
            }

            @Override
            public synchronized void reset() {
                AbstractIoBuffer.this.reset();
            }

            @Override
            public long skip(long n) {
                int bytes;
                if (n > Integer.MAX_VALUE) {
                    bytes = AbstractIoBuffer.this.remaining();
                } else {
                    bytes = Math.min(AbstractIoBuffer.this.remaining(), (int) n);
                }
                AbstractIoBuffer.this.skip(bytes);
                return bytes;
            }
        };
    }