在前面的博文中,介紹一些消息分割的方案,以及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。
新聞熱點
疑難解答