1.環(huán)境搭建
首先，我們需要添加兩個包，分別是:
mina-core-2.0.16.jar、slf4j-android-1.6.1-RC1.jar
2.自定義消息格式
在這里，我們規(guī)定每條消息的格式為：消息頭（一個int類型：表示消息體的長度、一個short類型：表示事件號）+消息體（為字符串，可以假定認為是json字符串）。接下來的很多代碼都是以此消息格式為基礎(chǔ)編寫。
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ā)送個接收端。
斷包：當(dāng)消息長度過大，那么就可能會將其分片，然后每片被TCP封裝，然后由IP封裝，最后被傳輸?shù)浇邮斩恕?br> 當(dāng)出現(xiàn)斷包或粘包現(xiàn)象，接收端接收到消息后，就會不清楚這是不是一個完整的消息，所以必須提供拆包機制。更多關(guān)于Socket斷包粘包的信息可自行上網(wǎng)搜索，在這里就不一一細說。
3.界面及代碼流程
1.界面
很簡單，見名知意。
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è)置進去。當(dāng)客戶端接收到消息的時候，消息會首先經(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、當(dāng)內(nèi)容剛好時，返回false，告知父類接收下一批內(nèi)容
* 2、內(nèi)容不夠時需要下一批發(fā)過來的內(nèi)容，此時返回false，這樣父類 CumulativeProtocolDecoder
* 會將內(nèi)容放進IoSession中，等下次來數(shù)據(jù)后就自動拼裝再交給本類的doDecode
* 3、當(dāng)內(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)//是用來當(dāng)拆包時候剩余長度小于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.itdecent.cn/p/2501310b03d2
