美文网首页
Android之Mina集成,自定义消息断包、粘包处理

Android之Mina集成,自定义消息断包、粘包处理

作者: 加油码农 | 来源:发表于2018-01-07 19:08 被阅读37次

    1.环境搭建
    首先,我们需要添加两个包,分别是:
    mina-core-2.0.16.jar、slf4j-android-1.6.1-RC1.jar
    2.自定义消息格式
    在这里,我们规定每条消息的格式为:消息头(一个int类型:表示消息体的长度、一个short类型:表示事件号)+消息体(为字符串,可以假定认为是json字符串)。接下来的很多代码都是以此消息格式为基础编写。


    2669657-631ac2f49b9e867d.png
    public class MinaMsgHead {
        public short     event;//消息事件号 2位
        public int       bodyLen;//消息内容长度 4位
    }
    

    对于这个消息格式,可能有的人会不理解为什么消息头的bodyLen值要等于消息体的长度呢,主要是因为TCP是面向连接的,面向流的,它是无消息保护边界, 需要在消息接收端处理消息边界问题。
    粘包:发送端为了提高网络利用率,使用了优化方法(Nagle算法),将多次间隔较小且数据量小的数据,合并成一个大的数据块,然后进行封包再发送个接收端。

    断包:当消息长度过大,那么就可能会将其分片,然后每片被TCP封装,然后由IP封装,最后被传输到接收端。
    当出现断包或粘包现象,接收端接收到消息后,就会不清楚这是不是一个完整的消息,所以必须提供拆包机制。更多关于Socket断包粘包的信息可自行上网搜索,在这里就不一一细说。

    3.界面及代码流程
    1.界面
    很简单,见名知意。


    2669657-8dd8cfe3ef729510.png

    2.开启服务
    以下是服务类的代码。 服务一开启便创建一个线程去连接,直到连接成功,其中ConnectionManager是一个连接管理类,接下来会介绍到。

    public class CoreService extends Service {
        public static final String TAG="CoreService";
        private ConnectionThread thread;
        ConnectionManager mManager;
        public CoreService() {
        }
        @Override
        public void onCreate() {
            ConnectionConfig config = new ConnectionConfig.Builder(getApplicationContext())
                    .setIp("192.168.0.88")//连接的IP地址,填写自己的服务器
                    .setPort(8888)//连接的端口号,填写自己的端口号
                    .setReadBufferSize(1024)
                    .setConnectionTimeout(10000).builder();
            mManager = new ConnectionManager(config);
            thread = new ConnectionThread();
            thread.start();
            super.onCreate();
        }
        @Override
        public IBinder onBind(Intent intent) {
            return null;
        }
    
        @Override
        public int onStartCommand(Intent intent, int flags, int startId) {
            return super.onStartCommand(intent, flags, startId);
        }
    
        class ConnectionThread extends Thread {
            boolean isConnection;
            @Override
            public void run() {
                for (;;){
                    isConnection = mManager.connect();
                    if (isConnection) {
                        Log.d(TAG,"连接成功跳出循环");
                        break;
                    }
                    try {
                        Log.d(TAG,"尝试重新连接");
                        Thread.sleep(5000);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            }
        }
        public void disConnect() {
            mManager.disConnect();
        }
    
        @Override
        public void onDestroy() {
            super.onDestroy();
            disConnect();
            thread = null;
        }
    }
    

    3.ConnectionManager类

    在这里最主要的就是MyCodecFactory类和DefaultHandler类,前者是我们的自定义编解码工厂,后者是我们的消息处理类,在初始化ConnectionManager 的时候,我们把它俩都设置进去。当客户端接收到消息的时候,消息会首先经过MyCodecFactory类的解码操作,解码完成后会再调用DefaultHandler的messageReceived方法。

    public class ConnectionManager {
        public static final String TAG="ConnectionManager";
        private ConnectionConfig mConfig;
        private WeakReference<Context> mContext;
        public NioSocketConnector mConnection;
        private IoSession mSession;
        private InetSocketAddress mAddress;
        public ConnectionManager(ConnectionConfig config){
            this.mConfig = config;
            this.mContext = new WeakReference<Context>(config.getContext());
            init();
        }
    
        private void init() {
            mAddress = new InetSocketAddress(mConfig.getIp(), mConfig.getPort());
            mConnection = new NioSocketConnector();
            mConnection.getSessionConfig().setReadBufferSize(mConfig.getReadBufferSize());
            //设置多长时间没有进行读写操作进入空闲状态,会调用sessionIdle方法,单位(秒)
            mConnection.getSessionConfig().setReaderIdleTime(60*5);
            mConnection.getSessionConfig().setWriterIdleTime(60*5);
            mConnection.getSessionConfig().setBothIdleTime(60*5);
            mConnection.getFilterChain().addFirst("reconnection", new MyIoFilterAdapter());
            //自定义编解码器
            mConnection.getFilterChain().addLast("mycoder", new ProtocolCodecFilter(new MyCodecFactory()));
            //添加消息处理器
            mConnection.setHandler(new DefaultHandler(mContext.get()));
            mConnection.setDefaultRemoteAddress(mAddress);
        }
    
        /**
         * 与服务器连接
         * @return true连接成功,false连接失败
         */
        public boolean connect(){
            try{
                ConnectFuture future = mConnection.connect();
                future.awaitUninterruptibly();
                mSession = future.getSession();
                if(mSession!=null && mSession.isConnected()) {
                    SessionManager.getInstance().setSession(mSession);
                }else {
                    return false;
                }
            }catch (Exception e){
                e.printStackTrace();
                return false;
            }
            return true;
        }
    
        /**
         * 断开连接
         */
        public void disConnect(){
            mConnection.dispose();
            mConnection=null;
            mSession=null;
            mAddress=null;
            mContext = null;
        }
        ...
    }
    

    4.MyCodecFactory类
    这只是一个实现ProtocolCodecFactory接口的工厂类,返回我们自定义编码和解码类。接下来的MyDataDecoder类就是我们的重点。

    public class MyCodecFactory implements ProtocolCodecFactory {
        private MyDataDecoder decoder;
        private MyDataEncoder encoder;
        public MyCodecFactory() {
            encoder = new MyDataEncoder();
            decoder = new MyDataDecoder();
        }
        @Override
        public ProtocolDecoder getDecoder(IoSession session) throws Exception {
            return decoder;
        }
        @Override
        public ProtocolEncoder getEncoder(IoSession session) throws Exception {
            return encoder;
        }
    }
    

    5.MyDataDecoder类
    以上说到“消息会首先经过MyCodecFactory类的解码操作”,其意为首先会经过MyDataDecoder 类的doDecode方法,这个方法的返回值是非常重要的(见注释),这个方法主要就是处理断包和粘包,只有收到的是完整的数据才交给我们的消息处理类DefaultHandler进行下一步的解析。

    public class MyDataDecoder extends CumulativeProtocolDecoder {
        /**
         * 返回值含义:
         * 1、当内容刚好时,返回false,告知父类接收下一批内容
         * 2、内容不够时需要下一批发过来的内容,此时返回false,这样父类 CumulativeProtocolDecoder
         * 会将内容放进IoSession中,等下次来数据后就自动拼装再交给本类的doDecode
         * 3、当内容多时,返回true,因为需要再将本批数据进行读取,父类会将剩余的数据再次推送本类的doDecode方法
         */
        @Override
        public boolean doDecode(IoSession session, IoBuffer in, ProtocolDecoderOutput out)
                throws Exception {
            /**
             * 假定消息格式为:消息头(int类型:表示消息体的长度、short类型:表示事件号)+消息体
             */
            if (in.remaining() < 4)//是用来当拆包时候剩余长度小于4的时候的保护,不加容易出错
            {
                return false;
            }
            if (in.remaining() > 1) {
                //以便后继的reset操作能恢复position位置
                in.mark();
                //前6字节是包头,一个int和一个short,我们先取一个int
                int len = in.getInt();//先获取包体数据长度值
    
                //比较消息长度和实际收到的长度是否相等,这里-2是因为我们的消息头有个short值还去取
                if (len > in.remaining() - 2) {
                    //出现断包,则重置恢复position位置到操作前,进入下一轮, 接收新数据,以拼凑成完整数据
                    in.reset();
                    return false;
                } else {
                    //消息内容足够
                    in.reset();//重置恢复position位置到操作前
                    int sumLen = 6 + len;//总长 = 包头+包体
                    byte[] packArr = new byte[sumLen];
                    in.get(packArr, 0, sumLen);
                    IoBuffer buffer = IoBuffer.allocate(sumLen);
                    buffer.put(packArr);
                    buffer.flip();
                    out.write(buffer);
                    //走到这里会调用DefaultHandler的messageReceived方法
    
                    if (in.remaining() > 0) {//出现粘包,就让父类再调用一次,进行下一次解析
                        return true;
                    }
                }
            }
            return false;//处理成功,让父类进行接收下个包
        }
    }
    

    6.DefaultHandler 类和HandlerEvent 类

    DefaultHandler 为我们消息处理类,经过了以上doDecode方法的断包和粘包处理,现在一个包头(int,short)+包体格式的消息完整来到了这里,在messageReceived要怎么解析数据就是你的事了,HandlerEvent 我是写了一个例子:依据包头的short值做为事件号处理。

    private class DefaultHandler extends IoHandlerAdapter {
            private Context mContext;
            private DefaultHandler(Context context){
                this.mContext = context;
            }
    
            @Override
            public void sessionOpened(IoSession session) throws Exception {
                super.sessionOpened(session);
                Log.d(TAG, "连接打开");
            }
    
            @Override
            public void messageReceived(IoSession session, Object message) throws Exception {
                Log.d(TAG, "收到数据,接下来你要怎么解析数据就是你的事了");
                IoBuffer buf = (IoBuffer) message;
                HandlerEvent.getInstance().handle(buf);
            }
    
            @Override
            public void sessionIdle(IoSession session, IdleStatus status) throws Exception {
                super.sessionIdle(session, status);
                Log.d(TAG, "-客户端与服务端连接空闲");
                //进入空闲状态我们把会话关闭,接着会调用MyIoFilterAdapter的sessionClosed方法,进行重新连接
                if(session != null){
                    session.closeOnFlush();
                }
            }
        }
    
    /**
     * 消息事件处理
     */
    public class HandlerEvent {
        private static HandlerEvent handlerEvent;
        public static HandlerEvent getInstance() {
            if (handlerEvent == null) {
                handlerEvent = new HandlerEvent();
            }
            return handlerEvent;
        }
        public void handle(IoBuffer buf) throws IOException, InterruptedException, UnsupportedEncodingException, SQLException {
            //解析包头
            MinaMsgHead msgHead = new MinaMsgHead();
            msgHead.bodyLen = buf.getInt();//包体长度
            msgHead.event = buf.getShort();//事件号
    
            switch (msgHead.event){//根据事件号解析消息体内容
                case Event.EV_S_C_TEST:
                    byte[] by = new byte[msgHead.bodyLen];
                    buf.get(by, 0, by.length);
                    String json = new String(by, "UTF-8").trim();
                    //接下来根据业务处理....
                    break;
            }
        }
    }
    

    总结
    在这个Demo里只是我的一个自定义消息格式,不同的消息格式在进行断包和粘包处理的写法有所不同,但都是万变不离其中,大家可以自行尝试。如有不足之处,欢迎指正。有需要Demo源码的同学请点击源码
    文章转载自:https://www.jianshu.com/p/2501310b03d2

    相关文章

      网友评论

          本文标题:Android之Mina集成,自定义消息断包、粘包处理

          本文链接:https://www.haomeiwen.com/subject/ctjmnxtx.html