這篇文章給大家分享的是有關Netty、MINA、Twisted中如何定制自己的協議的內容。小編覺得挺實用的，因此分享給大家做個參考，一起跟隨小編過來看看吧。
之前介紹了一些消息分割的方案，以及 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實現編碼解碼。雖然實現方式不同，但是它們的功能都是一樣的：
對消息根據一定規則進行切合，例如固定長度消息、按行、按分隔符、固定長度Header指定Body長度等；
將切合后的消息由字節碼轉成自己想要的類型，如MINA中將IoBuffer轉成字符串，這樣messageReceived接收到的message參數就是String類型；
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:6
Body:收到
由于一個漢字一般占3個字節，所以兩個漢字對應的Header為6。
感謝各位的閱讀！關于“Netty、MINA、Twisted中如何定制自己的協議”這篇文章就分享到這里了，希望以上內容可以對大家有一定的幫助，讓大家可以學到更多知識，如果覺得文章不錯，可以把它分享出去讓更多的人看到吧！
免責聲明：本站發布的內容（圖片、視頻和文字）以原創、轉載和分享為主，文章觀點不代表本網站立場，如果涉及侵權請聯系站長郵箱：is@yisu.com進行舉報，并提供相關證據，一經查實，將立刻刪除涉嫌侵權內容。
