登录 注册

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

本贴最后更新于 2167 天前，其中的信息可能已经时异事殊

1.环境搭建

  • 首先，我们需要添加两个包，分别是:
    mina-core-2.0.16.jar、slf4j-android-1.6.1-RC1.jar

2.自定义消息格式

  • 在这里，我们规定每条消息的格式为：消息头（一个 int 类型：表示消息体的长度、一个 short 类型：表示事件号）+ 消息体（为字符串，可以假定认为是 json 字符串）。接下来的很多代码都是以此消息格式为基础编写。


_

消息格式.png

_

public class MinaMsgHead {
    public short     event;//消息事件号 2位
    public int       bodyLen;//消息内容长度 4位
}

对于这个消息格式，可能有的人会不理解为什么消息头的 bodyLen 值要等于消息体的长度呢，主要是因为 TCP 是面向连接的，面向流的，它是无消息保护边界, 需要在消息接收端处理消息边界问题。

  • **粘包：**发送端为了提高网络利用率，使用了优化方法（Nagle 算法），将多次间隔较小且数据量小的数据，合并成一个大的数据块，然后进行封包再发送个接收端。
  • 断包：当消息长度过大，那么就可能会将其分片，然后每片被 TCP 封装，然后由 IP 封装，最后被传输到接收端。

当出现断包或粘包现象，接收端接收到消息后，就会不清楚这是不是一个完整的消息，所以必须提供拆包机制。更多关于 Socket 断包粘包的信息可自行上网搜索，在这里就不一一细说。

3.界面及代码流程

1.界面

  • 很简单，见名知意。


    _

    界面

    _

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 源码的同学请点击 ☞ 源码下载

  • MINA
    3 引用
  • Android

    Android 是一种以 Linux 为基础的开放源码操作系统，主要使用于便携设备。2005 年由 Google 收购注资，并拉拢多家制造商组成开放手机联盟开发改良，逐渐扩展到到平板电脑及其他领域上。

    334 引用 • 323 回帖 • 2 关注
  • Java

    Java 是一种可以撰写跨平台应用软件的面向对象的程序设计语言，是由 Sun Microsystems 公司于 1995 年 5 月推出的。Java 技术具有卓越的通用性、高效性、平台移植性和安全性。

    3187 引用 • 8213 回帖
685 14 323 67 19 27 2 38 181

相关帖子

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

欢迎来到这里！

我们正在构建一个小众社区，大家在这里相互信任，以平等 • 自由 • 奔放的价值观进行分享交流。最终，希望大家能够找到与自己志同道合的伙伴，共同成长。

注册 关于
请输入回帖内容 ...
woyehua
回帖
0
 帖子
16
 积分
540