在前面的博文中，介紹一些消息分割的方案，以及MINA、Netty、Twisted針對這些方案提供的相關API。例如MINA的TextLineCodecFactory、PRefixedStringCodecFactory，Netty的LineBasedFrameDecoder、LengthFieldBasedFrameDecoder，Twisted的LineOnlyReceiver、Int32StringReceiver。

除了這些方案，還有很多其他方案，當然也可以自己定義。在這里，我們定制一個自己的方案，并分別使用MINA、Netty、Twisted實現對這種消息的解析和組裝，也就是編碼和解碼。

上一篇博文中介紹了一種用固定字節數的Header來指定Body字節數的消息分割方案，其中Header部分是常規的大字節序（Big-Endian）的4字節整數。本文中對這個方案稍作修改，將固定字節數的Header改為小字節序（Little-Endian）的4字節整數。

常規的大字節序表示一個數的話，用高字節位的存放數字的低位，比較符合人的習慣。而小字節序和大字節序正好相反，用高字節位存放數字的高位。

Python中struct模塊支持大小字節序的pack和unpack，在java中可以用下面的兩個方法實現小字節序字節數組轉int和int轉小字節序字節數組，下面的Java程序中將會用到這兩個方法：

public class LittleEndian {            /**      * 將int轉成4字節的小字節序字節數組      */      public static byte[] toLittleEndian(int i) {          byte[] bytes = new byte[4];          bytes[0] = (byte) i;          bytes[1] = (byte) (i >>> 8);          bytes[2] = (byte) (i >>> 16);          bytes[3] = (byte) (i >>> 24);          return bytes;      }            /**      * 將小字節序的4字節的字節數組轉成int      */      public static int getLittleEndianInt(byte[] bytes) {          int b0 = bytes[0] & 0xFF;          int b1 = bytes[1] & 0xFF;          int b2 = bytes[2] & 0xFF;          int b3 = bytes[3] & 0xFF;          return b0 + (b1 << 8) + (b2 << 16) + (b3 << 24);      }  }  

無論是MINA、Netty還是Twisted，消息的編碼、解碼、切合的代碼，都是應該和業務邏輯代碼分開，這樣有利于代碼的開發、重用和維護。在MINA和Netty中類似，編碼、解碼需要繼承實現相應的Encoder、Decoder，而在Twisted中則是繼承Protocol實現編碼解碼。雖然實現方式不同，但是它們的功能都是一樣的：

1、對消息根據一定規則進行切合，例如固定長度消息、按行、按分隔符、固定長度Header指定Body長度等；

2、將切合后的消息由字節碼轉成自己想要的類型，如MINA中將IoBuffer轉成字符串，這樣messageReceived接收到的message參數就是String類型；

3、write的時候可以傳入自定義類型的參數，由編碼器完成編碼。

下面分別用MINA、Netty、Twisted實現4字節的小字節序int來指定body長度的消息的編碼解碼。

MINA：

在MINA中對接收到的消息進行切合和解碼，一般會定義一個解碼器類，繼承自抽象類CumulativeProtocolDecoder，實現doDecode方法：

public class MyMinaDecoder extends CumulativeProtocolDecoder {        @Override      protected boolean doDecode(Iosession session, IoBuffer in, ProtocolDecoderOutput out) throws Exception {                    // 如果沒有接收完Header部分（4字節），直接返回false          if(in.remaining() < 4) {              return false;          } else {                            // 標記開始位置，如果一條消息沒傳輸完成則返回到這個位置              in.mark();                            byte[] bytes = new byte[4];              in.get(bytes); // 讀取4字節的Header                            int bodyLength = LittleEndian.getLittleEndianInt(bytes); // 按小字節序轉int                            // 如果body沒有接收完整，直接返回false              if(in.remaining() < bodyLength) {                  in.reset(); // IoBuffer position回到原來標記的地方                  return false;              } else {                  byte[] bodyBytes = new byte[bodyLength];                  in.get(bodyBytes);                  String body = new String(bodyBytes, "UTF-8");                  out.write(body); // 解析出一條消息                  return true;              }          }      }  }  

另外，session.write的時候要對數據編碼，需要定義一個編碼器，繼承自抽象類ProtocolEncoderAdapter，實現encode方法：

public class MyMinaEncoder extends ProtocolEncoderAdapter {        @Override      public void encode(IoSession session, Object message,              ProtocolEncoderOutput out) throws Exception {                    String msg = (String) message;          byte[] bytes = msg.getBytes("UTF-8");          int length = bytes.length;          byte[] header = LittleEndian.toLittleEndian(length); // 按小字節序轉成字節數組                    IoBuffer buffer = IoBuffer.allocate(length + 4);          buffer.put(header); // header          buffer.put(bytes); // body          buffer.flip();          out.write(buffer);      }  }  

在服務器啟動的時候加入相應的編碼器和解碼器：

public class TcpServer {        public static void main(String[] args) throws IOException {          IoAcceptor acceptor = new NioSocketAcceptor();            // 指定編碼解碼器          acceptor.getFilterChain().addLast("codec",                   new ProtocolCodecFilter(new MyMinaEncoder(), new MyMinaDecoder()));            acceptor.setHandler(new TcpServerHandle());          acceptor.bind(new InetSocketAddress(8080));      }  }  

下面是業務邏輯的代碼：

public class TcpServerHandle extends IoHandlerAdapter {        @Override      public void exceptionCaught(IoSession session, Throwable cause)              throws Exception {          cause.printStackTrace();      }        // 接收到新的數據      @Override      public void messageReceived(IoSession session, Object message)              throws Exception {            // MyMinaDecoder將接收到的數據由IoBuffer轉為String          String msg = (String) message;          System.out.println("messageReceived:" + msg);                    // MyMinaEncoder將write的字符串添加了一個小字節序Header并轉為字節碼          session.write("收到");      }  }  

Netty：

Netty中解碼器和MINA類似，解碼器繼承抽象類ByteToMessageDecoder，實現decode方法：

public class MyNettyDecoder extends ByteToMessageDecoder {        @Override      protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {                    // 如果沒有接收完Header部分（4字節），直接退出該方法          if(in.readableBytes() >= 4) {                            // 標記開始位置，如果一條消息沒傳輸完成則返回到這個位置              in.markReaderIndex();                            byte[] bytes = new byte[4];              in.readBytes(bytes); // 讀取4字節的Header                            int bodyLength = LittleEndian.getLittleEndianInt(bytes); // header按小字節序轉int                            // 如果body沒有接收完整              if(in.readableBytes() < bodyLength) {                  in.resetReaderIndex(); // ByteBuf回到標記位置              } else {                  byte[] bodyBytes = new byte[bodyLength];                  in.readBytes(bodyBytes);                  String body = new String(bodyBytes, "UTF-8");                  out.add(body); // 解析出一條消息              }          }      }  }  

下面是編碼器，繼承自抽象類MessageToByteEncoder，實現encode方法：

public class MyNettyEncoder extends MessageToByteEncoder<String> {        @Override      protected void encode(ChannelHandlerContext ctx, String msg, ByteBuf out)              throws Exception {                    byte[] bytes = msg.getBytes("UTF-8");          int length = bytes.length;          byte[] header = LittleEndian.toLittleEndian(length); // int按小字節序轉字節數組          out.writeBytes(header); // write header          out.writeBytes(bytes); // write body      }  }  

加上相應的編碼器和解碼器：

public class TcpServer {        public static void main(String[] args) throws InterruptedException {          EventLoopGroup bossGroup = new NioEventLoopGroup();          EventLoopGroup workerGroup = new NioEventLoopGroup();          try {              ServerBootstrap b = new ServerBootstrap();              b.group(bossGroup, workerGroup)                      .channel(NioServerSocketChannel.class)                      .childHandler(new ChannelInitializer<SocketChannel>() {                          @Override                          public void initChannel(SocketChannel ch)                                  throws Exception {                              ChannelPipeline pipeline = ch.pipeline();                                // 加上自己的Encoder和Decoder                              pipeline.addLast(new MyNettyDecoder());                              pipeline.addLast(new MyNettyEncoder());                                pipeline.addLast(new TcpServerHandler());                          }                      });              ChannelFuture f = b.bind(8080).sync();              f.channel().closeFuture().sync();          } finally {              workerGroup.shutdownGracefully();              bossGroup.shutdownGracefully();          }      }  }  

業務邏輯處理類：

public class TcpServerHandler extends ChannelInboundHandlerAdapter {        // 接收到新的數據      @Override      public void channelRead(ChannelHandlerContext ctx, Object msg) {            // MyNettyDecoder將接收到的數據由ByteBuf轉為String          String message = (String) msg;          System.out.println("channelRead:" + message);            // MyNettyEncoder將write的字符串添加了一個小字節序Header并轉為字節碼          ctx.writeAndFlush("收到");      }        @Override      public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {          cause.printStackTrace();          ctx.close();      }  }  

Twisted：

Twisted的實現方式和MINA、Netty不太一樣，其實現方式相對來說更加原始，但是越原始也越接近底層原理。

首先要定義一個MyProtocol類繼承自Protocol，用于充當類似于MINA、Netty的編碼、解碼器。處理業務邏輯的類TcpServerHandle繼承MyProtocol，重寫或調用MyProtocol提供的一些方法。

# -*- coding:utf-8 –*-    from struct import pack, unpack  from twisted.internet.protocol import Factory  from twisted.internet.protocol import Protocol  from twisted.internet import reactor    # 編碼、解碼器  class MyProtocol(Protocol):            # 用于暫時存放接收到的數據      _buffer = b""            def dataReceived(self, data):          # 上次未處理的數據加上本次接收到的數據          self._buffer = self._buffer + data          # 一直循環直到新的消息沒有接收完整          while True:              # 如果header接收完整              if len(self._buffer) >= 4:                  # 按小字節序轉int                  length, = unpack("<I", self._buffer[0:4])                  # 如果body接收完整                  if len(self._buffer) >= 4 + length:                      # body部分                      packet = self._buffer[4:4 + length]                      # 新的一條消息接收并解碼完成，調用stringReceived                      self.stringReceived(packet)                      # 去掉_buffer中已經處理的消息部分                      self._buffer = self._buffer[4 + length:]                  else:                      break;              else:                  break;                    def stringReceived(self, data):          raise NotImplementedError            def sendString(self, string):          self.transport.write(pack("<I", len(string)) + string)    # 邏輯代碼  class TcpServerHandle(MyProtocol):        # 實現MyProtocol提供的stringReceived而不是dataReceived，不然無法解碼      def stringReceived(self, data):                    # data為MyProtocol解碼后的數據          print 'stringReceived:' + data                    # 調用sendString而不是self.transport.write，不然不能進行編碼          self.sendString("收到")    factory = Factory()  factory.protocol = TcpServerHandle  reactor.listenTCP(8080, factory)  reactor.run()  

下面是Java編寫的一個客戶端測試程序：

public class TcpClient {        public static void main(String[] args) throws IOException {            Socket socket = null;          OutputStream out = null;          InputStream in = null;                    try {                socket = new Socket("localhost", 8080);              out = socket.getOutputStream();              in = socket.getInputStream();                            // 請求服務器              String data = "我是客戶端";              byte[] outputBytes = data.getBytes("UTF-8");              out.write(LittleEndian.toLittleEndian(outputBytes.length)); // write header              out.write(outputBytes); // write body              out.flush();                            // 獲取響應              byte[] inputBytes = new byte[1024];              int length = in.read(inputBytes);              if(length >= 4) {                  int bodyLength = LittleEndian.getLittleEndianInt(inputBytes);                  if(length >= 4 + bodyLength) {                      byte[] bodyBytes = Arrays.copyOfRange(inputBytes, 4, 4 + bodyLength);                      System.out.println("Header:" + bodyLength);                      System.out.println("Body:" + new String(bodyBytes, "UTf-8"));                  }              }            } finally {              // 關閉連接              in.close();              out.close();              socket.close();          }      }  }  

用客戶端分別測試上面三個TCP服務器：

MINA服務器輸出結果：

messageReceived:我是客戶端

Netty服務器輸出結果：

channelRead:我是客戶端

Twisted服務器輸出結果：

stringReceived:我是客戶端

客戶端測試三個服務器的輸出結果都是：

Header:6Body:收到

由于一個漢字一般占3個字節，所以兩個漢字對應的Header為6。