Mina 编写自己的编码解码FilterChain(codec)

mina 自己带的FilterChain codec是非常好用，但是在实际网络应用中还是有他的局限性，如编写基于CMPP、SGIP的短信系统。

下面我编写的一个自己FilterChain例子，方便以后查看和其他的人查阅（哎，网上的mina资料少的可怜）

1.mina服务器类

package Minatest;

import java.net.InetSocketAddress;

import org.apache.mina.common.DefaultIoFilterChainBuilder;
import org.apache.mina.common.IdleStatus;
import org.apache.mina.common.IoAcceptor;
import org.apache.mina.common.IoAcceptorConfig;
import org.apache.mina.common.IoHandler;
import org.apache.mina.common.IoSession;
import org.apache.mina.filter.LoggingFilter;
import org.apache.mina.filter.codec.ProtocolCodecFilter;
import org.apache.mina.transport.socket.nio.SocketAcceptor;
import org.apache.mina.transport.socket.nio.SocketAcceptorConfig;

public class MinaServer {

/**
* @param args
* @throws Exception
* zxp
*
*/
public static void main(String[] args) throws Exception {
   IoAcceptor acceptor=new SocketAcceptor();
   IoAcceptorConfig config=new SocketAcceptorConfig();
   //config.setDisconnectOnUnbind(true);
   DefaultIoFilterChainBuilder d= config.getFilterChain();
   //d.addLast("codec", new ProtocolCodecFilter(new TextLineCodecFactory( Charset.forName("UTF-8"))));
   //d.addFirst("a", );
   InetSocketAddress address=new InetSocketAddress(9000);
   acceptor.bind(address, new MinaServer().new server(), config);
 
}
public class server implements IoHandler {

   public void exceptionCaught(IoSession arg0, Throwable arg1) throws Exception {
    // TODO Auto-generated method stub
    System.out.println("异常:"+arg1);
    arg1.fillInStackTrace();
   }

   public void messageReceived(IoSession arg0, Object arg1) throws Exception {
    // TODO Auto-generated method stub
    System.out.println("收到信息");
    SendMessages send=(SendMessages)arg1;
    System.out.println(send.msg);
    //收到客户端的消息 为0表示刚刚发送的测试消息发送成功 （这个流程是针对CMPP协议，对于普通的来说没意义）
    if(send.msg.equals("0")){
     this.sessionClosed(arg0);
    }
  
  
   }
        //信息发送成功后触发的事件
   public void messageSent(IoSession arg0, Object arg1) throws Exception {
    // TODO Auto-generated method stub
    System.out.println("发送信息");
   }

   public void sessionClosed(IoSession arg0) throws Exception {
    // TODO Auto-generated method stub
    System.out.println("session 关闭");
    arg0.close();
   }

   public void sessionCreated(IoSession arg0) throws Exception {
    // TODO Auto-generated method stub
    System.out.println("session 建立");
   }

   public void sessionIdle(IoSession arg0, IdleStatus arg1) throws Exception {
    // TODO Auto-generated method stub
    System.out.println("session 空闲");
    if(arg1==IdleStatus.BOTH_IDLE){
//     //自定义发送类
//     SendMessages send= new SendMessages();
//     send.setMsg("Hi!客户端你空闲咯！");
//     this.sendMessage(arg0, send);
    }
   }

   public void sessionOpened(IoSession arg0) throws Exception {
    // TODO Auto-generated method stub
    System.out.println("session 打开");
  
    //arg0.setIdleTime(IdleStatus.BOTH_IDLE, 10);
    arg0.getFilterChain().addFirst("a", new ProtocolCodecFilter(new TestProtocolCodecFactory()));
    arg0.getFilterChain().addLast("logger", new LoggingFilter());
  
    //发送条测试消息
    SendMessages send= new SendMessages();
    send.setMsg("Hi!客户端你好！");
    this.sendMessage(arg0, send);
  
   }
 
   private void sendMessage(IoSession arg0, Object arg1){
    arg0.write(arg1);
   }
}

}

2.mina客户端类

package Minatest;

import java.net.InetSocketAddress;
import java.nio.charset.Charset;

import org.apache.mina.common.DefaultIoFilterChainBuilder;
import org.apache.mina.common.IdleStatus;
import org.apache.mina.common.IoConnector;
import org.apache.mina.common.IoConnectorConfig;
import org.apache.mina.common.IoHandler;
import org.apache.mina.common.IoSession;
import org.apache.mina.filter.codec.ProtocolCodecFactory;
import org.apache.mina.filter.codec.ProtocolCodecFilter;
import org.apache.mina.filter.codec.textline.TextLineCodecFactory;
import org.apache.mina.transport.socket.nio.SocketConnector;
import org.apache.mina.transport.socket.nio.SocketConnectorConfig;


public class MinaClient {

/**
* @param args
* zxp
*
*/
public static void main(String[] args) {
   // TODO Auto-generated method stub
       IoConnector connector=new SocketConnector();
       IoConnectorConfig config=new SocketConnectorConfig();
      //config.setConnectTimeout(1);
       //DefaultIoFilterChainBuilder d= config.getFilterChain();
       //d.addLast("codec", new ProtocolCodecFilter(new TextLineCodecFactory(Charset.forName("UTF-8"))));
       InetSocketAddress address=new InetSocketAddress("127.0.0.1",9000);
       connector.connect(address, new MinaClient().new client(), config);
      
}
   public class client implements IoHandler{

public void exceptionCaught(IoSession arg0, Throwable arg1) throws Exception {
   // TODO Auto-generated method stub
   System.out.println("客户端异常:"+arg1);
   arg0.close();
}

public void messageReceived(IoSession arg0, Object arg1) throws Exception {
   // TODO Auto-generated method stub
   System.out.println("客户端收到消息");
   SendMessages send=(SendMessages) arg1;
   System.out.println(send.getMsg());
   if(send.msg!=null){
    send.setMsg("0");
    this.sendMessages(arg0,send);
   }
}

public void messageSent(IoSession arg0, Object arg1) throws Exception {
   // TODO Auto-generated method stub
       //不能用这个方法发送信息，要不死循环
   //arg0.getw
}

public void sessionClosed(IoSession arg0) throws Exception {
   // TODO Auto-generated method stub
   System.out.println("客户端关闭");
   arg0.close();
}

public void sessionCreated(IoSession arg0) throws Exception {
   // TODO Auto-generated method stub
   System.out.println("客户端建立");
   //arg0.getFilterChain()
}

public void sessionIdle(IoSession arg0, IdleStatus arg1) throws Exception {
   // TODO Auto-generated method stub
   System.out.println("客户端空闲");
   if(arg1==IdleStatus.BOTH_IDLE){
    this.messageSent(arg0, "空闲测试");
    SendMessages send=new SendMessages();
    send.setMsg("aaaaa");
    this.sendMessages(arg0, send);
   }
}

public void sessionOpened(IoSession arg0) throws Exception {
   // TODO Auto-generated method stub
   System.out.println("客户端打开");
   //10秒内没有读写就设置为空闲通道
   arg0.setIdleTime(IdleStatus.BOTH_IDLE, 10);
 
   //自定义包解析
     //ProtocolCodecFactory codec=new TestProtocolCodecFactory();
   //arg0.getFilterChain().addFirst("test", new ProtocolCodecFilter(codec));
     arg0.getFilterChain().addLast("a", new ProtocolCodecFilter(new TestProtocolCodecFactory()));
}
   
private void sendMessages(IoSession arg0, Object arg1){
   System.out.println("sendMessages 发送");
   arg0.write(arg1);
}

   }
}

3.工厂类

package Minatest;

import org.apache.mina.filter.codec.demux.DemuxingProtocolCodecFactory;

public class TestProtocolCodecFactory extends DemuxingProtocolCodecFactory {
   public TestProtocolCodecFactory(){
    super.register(TestDecoder.class);
    super.register(TestEncoder.class);
   }
}

4.编码类

package Minatest;

import java.util.Collections;
import java.util.HashSet;
import java.util.Hashtable;
import java.util.Set;

import org.apache.mina.common.ByteBuffer;
import org.apache.mina.common.IoSession;
import org.apache.mina.filter.codec.ProtocolEncoderOutput;
import org.apache.mina.filter.codec.demux.MessageEncoder;


public class TestEncoder implements MessageEncoder {
/**
* 编码消息
* zxp
*
*/
public void encode(IoSession session, Object message, ProtocolEncoderOutput out) throws Exception {
   SendMessages msg=(SendMessages)message;
   ByteBuffer b=ByteBuffer.allocate(msg.getmsgleng());
   b.setAutoExpand(true);
   //编码消息
   msg.encoder(b);
   //b.putInt(97);
   //b.put("a".getBytes());
   b.flip();
   out.write(b);
}
/**
* 获取编码自定义消息类型(多个)
*
*/
public Set<Class<?>> getMessageTypes() {
   Set<Class<?>> set=new HashSet<Class<?>>();
   set.add(SendMessages.class);
   //返回指定映射的不可修改视图 即只读
   return Collections.unmodifiableSet(set);
}

}
5.解码类

package Minatest;

import org.apache.mina.common.ByteBuffer;
import org.apache.mina.common.IoSession;
import org.apache.mina.filter.codec.ProtocolDecoderOutput;
import org.apache.mina.filter.codec.demux.MessageDecoder;
import org.apache.mina.filter.codec.demux.MessageDecoderResult;

public class TestDecoder implements MessageDecoder {
/**
* 检查指定的缓冲区decodable解码器
* zxp
*/
public MessageDecoderResult decodable(IoSession arg0, ByteBuffer arg1) {
   //包的长度是否大于0
   if(arg1.remaining()<=0){
    return MessageDecoderResult.NOT_OK;
   }
   return MessageDecoderResult.OK;
}
/**
* 解码二进制或协议的具体内容到更高级别的消息对象
* (non-Javadoc)
* @see org.apache.mina.filter.codec.demux.MessageDecoder#decode(org.apache.mina.common.IoSession, org.apache.mina.common.ByteBuffer, org.apache.mina.filter.codec.ProtocolDecoderOutput)
*/
public MessageDecoderResult decode(IoSession arg0, ByteBuffer arg1,
    ProtocolDecoderOutput arg2) throws Exception {
    //解码
   byte[] b=new byte[arg1.limit()];
   arg1.get(b);
   System.out.println("decode:"+arg1.position());
   SendMessages send=new SendMessages();
   String s=new String(b);
   send.setMsg(s);
   arg2.write(send);
   return MessageDecoderResult.OK;
}

/**
* 解码完成后调用的方法
*
*/
public void finishDecode(IoSession arg0, ProtocolDecoderOutput arg1)
    throws Exception {

}

}

6.自定义消息类

package Minatest;

import org.apache.mina.common.ByteBuffer;

public class SendMessages implements java.io.Serializable {



    public String msg="0";

public String getMsg() {
   return msg;
}

public void setMsg(String msg) {
   this.msg = msg;
}

public int getmsgleng(){
   return msg.getBytes().length;
}

public void encoder(ByteBuffer b){
   b.put(msg.getBytes());
   //b.putInt(1);
}
   
}

评论

1 楼 yipanbo 2014-11-01  

博主好人！