Android之Mina集成饱普，自定義消息斷包骂删、粘包處理

1.環(huán)境搭建
首先，我們需要添加兩個包娇昙，分別是:
mina-core-2.0.16.jar尺迂、slf4j-android-1.6.1-RC1.jar
2.自定義消息格式
在這里，我們規(guī)定每條消息的格式為：消息頭（一個int類型：表示消息體的長度冒掌、一個short類型：表示事件號）+消息體（為字符串噪裕，可以假定認為是json字符串）。接下來的很多代碼都是以此消息格式為基礎(chǔ)編寫股毫。

2669657-631ac2f49b9e867d.png

public class MinaMsgHead {
    public short     event;//消息事件號 2位
    public int       bodyLen;//消息內(nèi)容長度 4位
}

對于這個消息格式膳音，可能有的人會不理解為什么消息頭的bodyLen值要等于消息體的長度呢，主要是因為TCP是面向連接的铃诬，面向流的祭陷，它是無消息保護邊界, 需要在消息接收端處理消息邊界問題。
粘包：發(fā)送端為了提高網(wǎng)絡(luò)利用率氧急，使用了優(yōu)化方法（Nagle算法）颗胡，將多次間隔較小且數(shù)據(jù)量小的數(shù)據(jù)，合并成一個大的數(shù)據(jù)塊吩坝，然后進行封包再發(fā)送個接收端毒姨。

斷包：當消息長度過大，那么就可能會將其分片钉寝，然后每片被TCP封裝弧呐，然后由IP封裝闸迷，最后被傳輸?shù)浇邮斩恕?br> 當出現(xiàn)斷包或粘包現(xiàn)象，接收端接收到消息后俘枫，就會不清楚這是不是一個完整的消息腥沽，所以必須提供拆包機制。更多關(guān)于Socket斷包粘包的信息可自行上網(wǎng)搜索鸠蚪，在這里就不一一細說今阳。

3.界面及代碼流程
1.界面
很簡單，見名知意茅信。

2669657-8dd8cfe3ef729510.png

2.開啟服務(wù)
以下是服務(wù)類的代碼盾舌。服務(wù)一開啟便創(chuàng)建一個線程去連接，直到連接成功蘸鲸，其中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地址酌摇，填寫自己的服務(wù)器
                .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,"連接成功跳出循環(huán)");
                    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 的時候，我們把它倆都設(shè)置進去怯伊。當客戶端接收到消息的時候琳轿，消息會首先經(jīng)過MyCodecFactory類的解碼操作，解碼完成后會再調(diào)用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());
        //設(shè)置多長時間沒有進行讀寫操作進入空閑狀態(tài)崭篡，會調(diào)用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);
    }

    /**
     * 與服務(wù)器連接
     * @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類
這只是一個實現(xiàn)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類
以上說到“消息會首先經(jīng)過MyCodecFactory類的解碼操作”颠毙，其意為首先會經(jīng)過MyDataDecoder 類的doDecode方法，這個方法的返回值是非常重要的（見注釋）砂碉，這個方法主要就是處理斷包和粘包蛀蜜，只有收到的是完整的數(shù)據(jù)才交給我們的消息處理類DefaultHandler進行下一步的解析。

public class MyDataDecoder extends CumulativeProtocolDecoder {
    /**
     * 返回值含義:
     * 1增蹭、當內(nèi)容剛好時滴某，返回false，告知父類接收下一批內(nèi)容
     * 2、內(nèi)容不夠時需要下一批發(fā)過來的內(nèi)容霎奢，此時返回false户誓，這樣父類 CumulativeProtocolDecoder
     * 會將內(nèi)容放進IoSession中，等下次來數(shù)據(jù)后就自動拼裝再交給本類的doDecode
     * 3幕侠、當內(nèi)容多時帝美，返回true，因為需要再將本批數(shù)據(jù)進行讀取晤硕，父類會將剩余的數(shù)據(jù)再次推送本類的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操作能恢復(fù)position位置
            in.mark();
            //前6字節(jié)是包頭舞箍，一個int和一個short女责，我們先取一個int
            int len = in.getInt();//先獲取包體數(shù)據(jù)長度值

            //比較消息長度和實際收到的長度是否相等，這里-2是因為我們的消息頭有個short值還去取
            if (len > in.remaining() - 2) {
                //出現(xiàn)斷包创译，則重置恢復(fù)position位置到操作前,進入下一輪, 接收新數(shù)據(jù)，以拼湊成完整數(shù)據(jù)
                in.reset();
                return false;
            } else {
                //消息內(nèi)容足夠
                in.reset();//重置恢復(fù)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);
                //走到這里會調(diào)用DefaultHandler的messageReceived方法

                if (in.remaining() > 0) {//出現(xiàn)粘包墙基，就讓父類再調(diào)用一次软族，進行下一次解析
                    return true;
                }
            }
        }
        return false;//處理成功，讓父類進行接收下個包
    }
}

6.DefaultHandler 類和HandlerEvent 類

DefaultHandler 為我們消息處理類残制，經(jīng)過了以上doDecode方法的斷包和粘包處理立砸，現(xiàn)在一個包頭（int，short）+包體格式的消息完整來到了這里初茶，在messageReceived要怎么解析數(shù)據(jù)就是你的事了颗祝，HandlerEvent 我是寫了一個例子：依據(jù)包頭的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, "收到數(shù)據(jù)恼布，接下來你要怎么解析數(shù)據(jù)就是你的事了");
            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, "-客戶端與服務(wù)端連接空閑");
            //進入空閑狀態(tài)我們把會話關(guān)閉螺戳，接著會調(diào)用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){//根據(jù)事件號解析消息體內(nèi)容
            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();
                //接下來根據(jù)業(yè)務(wù)處理....
                break;
        }
    }
}

總結(jié)
在這個Demo里只是我的一個自定義消息格式折汞，不同的消息格式在進行斷包和粘包處理的寫法有所不同倔幼，但都是萬變不離其中，大家可以自行嘗試爽待。如有不足之處损同，歡迎指正。有需要Demo源碼的同學(xué)請點擊源碼
文章轉(zhuǎn)載自：http://www.reibang.com/p/2501310b03d2

Android之Mina集成比被，自定義消息斷包、粘包處理

Android之Mina集成泼舱，自定義消息斷包等缀、粘包處理