Netty解決 TCP 粘包拆包的方法
什麼是粘包/拆包
一般所謂的TCP粘包是在一次接收數據不能完全地體現一個完整的消息數據。TCP通訊為何存在粘包呢？主要原因是TCP是以流的方式來處理數據，再加上網絡上MTU的往往小於在應用處理的消息數據，所以就會引發一次接收的數據無法滿足消息的需要，導致粘包的存在。處理粘包的唯一方法就是制定應用層的數據通訊協議，通過協議來規范現有接收的數據是否滿足消息數據的需要。
我們都知道TCP是基於字節流的傳輸協議。
那麼數據在通信層傳播其實就像河水一樣並沒有明顯的分界線，而數據具體表示什麼意思什麼地方有句號什麼地方有分號這個對於TCP底層來說並不清楚。應用層向TCP層發送用於網間傳輸的、用8位字節表示的數據流，然後TCP把數據流分區成適當長度的報文段，之後TCP把結果包傳給IP層，由它來通過網絡將包傳送給接收端實體的TCP層。
所以對於這個數據拆分成大包小包的問題就是我們今天要講的粘包和拆包的問題。
1、TCP粘包拆包問題說明
粘包和拆包這兩個概念估計大傢還不清楚，通過下面這張圖我們來分析一下：
假設客戶端分別發送兩個數據包D1,D2個服務端，但是發送過程中數據是何種形式進行傳播這個並不清楚，分別有下列4種情況：
- 服務端一次接受到瞭D1和D2兩個數據包，兩個包粘在一起，稱為粘包；
- 服務端分兩次讀取到數據包D1和D2，沒有發生粘包和拆包；
- 服務端分兩次讀到瞭數據包，第一次讀到瞭D1和D2的部分內容，第二次讀到瞭D2的剩下部分，這個稱為拆包；
- 服務器分三次讀到瞭數據部分，第一次讀到瞭D1包，第二次讀到瞭D2包的部分內容，第三次讀到瞭D2包的剩下內容。
2、TCP粘包產生原因
我們知道在TCP協議中，應用數據分割成TCP認為最適合發送的數據塊，這部分是通過“MSS”（最大數據包長度）選項來控制的，通常這種機制也被稱為一種協商機制，MSS規定瞭TCP傳往另一端的最大數據塊的長度。這個值TCP協議在實現的時候往往用MTU值代替（需要減去IP數據包包頭的大小20Bytes和TCP數據段的包頭20Bytes）所以往往MSS為1460。通訊雙方會根據雙方提供的MSS值得最小值確定為這次連接的最大MSS值。
tcp為提高性能，發送端會將需要發送的數據發送到緩沖區，等待緩沖區滿瞭之後，再將緩沖中的數據發送到接收方。同理，接收方也有緩沖區這樣的機制，來接收數據。
發生粘包拆包的原因主要有以下這些：
- 應用程序寫入數據的字節大小大於套接字發送緩沖區的大小將發生拆包；
- 進行MSS大小的TCP分段。MSS是TCP報文段中的數據字段的最大長度，當TCP報文長度-TCP頭部長度>mss的時候將發生拆包；
- 應用程序寫入數據小於套接字緩沖區大小，網卡將應用多次寫入的數據發送到網絡上,將發生粘包；
- 數據包大於MTU的時候將會進行切片。MTU即(Maxitum Transmission Unit) 最大傳輸單元,由於以太網傳輸電氣方面的限制，每個以太網幀都有最小的大小64bytes最大不能超過1518bytes,刨去以太網幀的幀頭14Bytes和幀尾CRC校驗部分4Bytes,那麼剩下承載上層協議的地方也就是Data域最大就隻能有1500Bytes這個值我們就把它稱之為MTU。這個就是網絡層協議非常關心的地方，因為網絡層協議比如IP協議會根據這個值來決定是否把上層傳下來的數據進行分片。
3、如何解決TCP粘包拆包
我們知道tcp是無界的數據流，且協議本身無法避免粘包，拆包的發生，那我們隻能在應用層數據協議上，加以控制。通常在制定傳輸數據時，可以使用如下方法：
- 設置定長消息，服務端每次讀取既定長度的內容作為一條完整消息；
- 使用帶消息頭的協議、消息頭存儲消息開始標識及消息長度信息，服務端獲取消息頭的時候解析出消息長度，然後向後讀取該長度的內容；
- 設置消息邊界，服務端從網絡流中按消息邊界分離出消息內容。比如在消息末尾加上換行符用以區分消息結束。
當然應用層還有更多復雜的方式可以解決這個問題，這個就屬於網絡層的問題瞭，我們還是用java提供的方式來解決這個問題。我們先看一個例子看看粘包是如何發生的。
服務端：
public class HelloWordServer {
private int port;
public HelloWordServer(int port) {
this.port = port;
}
public void start(){
EventLoopGroup bossGroup = new NioEventLoopGroup();
EventLoopGroup workGroup = new NioEventLoopGroup();
ServerBootstrap server = new ServerBootstrap().group(bossGroup,workGroup)
.channel(NioServerSocketChannel.class)
.childHandler(new ServerChannelInitializer());
try {
ChannelFuture future = server.bind(port).sync();
future.channel().closeFuture().sync();
} catch (InterruptedException e) {
e.printStackTrace();
}finally {
bossGroup.shutdownGracefully();
workGroup.shutdownGracefully();
}
}
public static void main(String[] args) {
HelloWordServer server = new HelloWordServer(7788);
server.start();
}
}
服務端Initializer：
public class ServerChannelInitializer extends ChannelInitializer<SocketChannel> {
@Override
protected void initChannel(SocketChannel socketChannel) throws Exception {
ChannelPipeline pipeline = socketChannel.pipeline();
// 字符串解碼 和 編碼
pipeline.addLast("decoder", new StringDecoder());
pipeline.addLast("encoder", new StringEncoder());
// 自己的邏輯Handler
pipeline.addLast("handler", new HelloWordServerHandler());
}
}
服務端handler：
public class HelloWordServerHandler extends ChannelInboundHandlerAdapter {
private int counter;
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
String body = (String)msg;
System.out.println("server receive order : " + body + ";the counter is: " + ++counter);
}
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
super.exceptionCaught(ctx, cause);
}
}
客戶端：
public class HelloWorldClient {
private int port;
private String address;
public HelloWorldClient(int port,String address) {
this.port = port;
this.address = address;
}
public void start(){
EventLoopGroup group = new NioEventLoopGroup();
Bootstrap bootstrap = new Bootstrap();
bootstrap.group(group)
.channel(NioSocketChannel.class)
.handler(new ClientChannelInitializer());
try {
ChannelFuture future = bootstrap.connect(address,port).sync();
future.channel().closeFuture().sync();
} catch (Exception e) {
e.printStackTrace();
}finally {
group.shutdownGracefully();
}
}
public static void main(String[] args) {
HelloWorldClient client = new HelloWorldClient(7788,"127.0.0.1");
client.start();
}
}
客戶端Initializer：
public class ClientChannelInitializer extends ChannelInitializer<SocketChannel> {
protected void initChannel(SocketChannel socketChannel) throws Exception {
ChannelPipeline pipeline = socketChannel.pipeline();
pipeline.addLast("decoder", new StringDecoder());
pipeline.addLast("encoder", new StringEncoder());
// 客戶端的邏輯
pipeline.addLast("handler", new HelloWorldClientHandler());
}
}
客戶端handler：
public class HelloWorldClientHandler extends ChannelInboundHandlerAdapter {
private byte[] req;
private int counter;
public BaseClientHandler() {
req = ("Unless required by applicable law or agreed to in writing, software\n" +
" distributed under the License is distributed on an \"AS IS\" BASIS,\n" +
" WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n" +
" See the License for the specific language governing permissions and\n" +
" limitations under the License.This connector uses the BIO implementation that requires the JSSE\n" +
" style configuration. When using the APR/native implementation, the\n" +
" penSSL style configuration is required as described in the APR/native\n" +
" documentation.An Engine represents the entry point (within Catalina) that processes\n" +
" every request. The Engine implementation for Tomcat stand alone\n" +
" analyzes the HTTP headers included with the request, and passes them\n" +
" on to the appropriate Host (virtual host)# Unless required by applicable law or agreed to in writing, software\n" +
"# distributed under the License is distributed on an \"AS IS\" BASIS,\n" +
"# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n" +
"# See the License for the specific language governing permissions and\n" +
"# limitations under the License.# For example, set the org.apache.catalina.util.LifecycleBase logger to log\n" +
"# each component that extends LifecycleBase changing state:\n" +
"#org.apache.catalina.util.LifecycleBase.level = FINE"
).getBytes();
}
@Override
public void channelActive(ChannelHandlerContext ctx) throws Exception {
ByteBuf message;
//將上面的所有字符串作為一個消息體發送出去
message = Unpooled.buffer(req.length);
message.writeBytes(req);
ctx.writeAndFlush(message);
}
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
String buf = (String)msg;
System.out.println("Now is : " + buf + " ; the counter is : "+ (++counter));
}
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
ctx.close();
}
}
運行客戶端和服務端我們能看到：
我們看到這個長長的字符串被截成瞭2段發送，這就是發生瞭拆包的現象。同樣粘包我們也很容易去模擬，我們把BaseClientHandler中的channelActive方法裡面的：
message = Unpooled.buffer(req.length); message.writeBytes(req); ctx.writeAndFlush(message);
這幾行代碼是把我們上面的一長串字符轉成的byte數組寫進流裡發送出去，那麼我們可以在這裡把上面發送消息的這幾行循環幾遍這樣發送的內容增多瞭就有可能在拆包的時候把上一條消息的一部分分配到下一條消息裡面瞭，修改如下：
for (int i = 0; i < 3; i++) {
message = Unpooled.buffer(req.length);
message.writeBytes(req);
ctx.writeAndFlush(message);
}
改完之後我們再運行一下，輸出太長不好截圖，我們在輸出結果中能看到循環3次之後的消息服務端收到的就不是之前的完整的一條瞭，而是被拆分瞭4次發送。
對於上面出現的粘包和拆包的問題，Netty已有考慮，並且有實施的方案：LineBasedFrameDecoder。
我們重新改寫一下ServerChannelInitializer：
public class ServerChannelInitializer extends ChannelInitializer<SocketChannel> {
@Override
protected void initChannel(SocketChannel socketChannel) throws Exception {
ChannelPipeline pipeline = socketChannel.pipeline();
pipeline.addLast(new LineBasedFrameDecoder(2048));
// 字符串解碼 和 編碼
pipeline.addLast("decoder", new StringDecoder());
pipeline.addLast("encoder", new StringEncoder());
// 自己的邏輯Handler
pipeline.addLast("handler", new BaseServerHandler());
}
}
新增：pipeline.addLast(new LineBasedFrameDecoder(2048))。同時，我們還得對上面發送的消息進行改造BaseClientHandler：
public class BaseClientHandler extends ChannelInboundHandlerAdapter {
private byte[] req;
private int counter;
req = ("Unless required by applicable dfslaw or agreed to in writing, software" +
" distributed under the License is distributed on an \"AS IS\" BASIS," +
" WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied." +
" See the License for the specific language governing permissions and" +
" limitations under the License.This connector uses the BIO implementation that requires the JSSE" +
" style configuration. When using the APR/native implementation, the" +
" penSSL style configuration is required as described in the APR/native" +
" documentation.An Engine represents the entry point (within Catalina) that processes" +
" every request. The Engine implementation for Tomcat stand alone" +
" analyzes the HTTP headers included with the request, and passes them" +
" on to the appropriate Host (virtual host)# Unless required by applicable law or agreed to in writing, software" +
"# distributed under the License is distributed on an \"AS IS\" BASIS," +
"# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied." +
"# See the License for the specific language governing permissions and" +
"# limitations under the License.# For example, set the org.apache.catalina.util.LifecycleBase logger to log" +
"# each component that extends LifecycleBase changing state:" +
"#org.apache.catalina.util.LifecycleBase.level = FINE\n"
).getBytes();
@Override
public void channelActive(ChannelHandlerContext ctx) throws Exception {
ByteBuf message;
message = Unpooled.buffer(req.length);
message.writeBytes(req);
ctx.writeAndFlush(message);
}
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
String buf = (String)msg;
System.out.println("Now is : " + buf + " ; the counter is : "+ (++counter));
}
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
ctx.close();
}
}
去掉所有的”\n”,隻保留字符串末尾的這一個。原因稍後再說。channelActive方法中我們不必再用循環多次發送消息瞭，隻發送一次就好(第一個例子中發送一次的時候是發生瞭拆包的)，然後我們再次運行，大傢會看到這麼長一串字符隻發送瞭一串就發送完畢。程序輸出我就不截圖瞭。下面來解釋一下LineBasedFrameDecoder。
LineBasedFrameDecoder的工作原理是它依次遍歷ByteBuf 中的可讀字節，判斷看是否有”\n” 或者” \r\n”，如果有，就以此位置為結束位置，從可讀索引到結束位置區間的字節就組成瞭一行。它是以換行符為結束標志的解碼器。支持攜帶結束符或者不攜帶結束符兩種解碼方式，同時支持配置單行的最大長度。如果連續讀取到最大長度後仍然沒有發現換行符，就會拋出異常，同時忽略掉之前讀到的異常碼流。這個對於我們確定消息最大長度的應用場景還是很有幫助。
對於上面的判斷看是否有”\n” 或者” \r\n”以此作為結束的標志我們可能回想，要是沒有”\n” 或者” \r\n”那還有什麼別的方式可以判斷消息是否結束呢。別擔心，Netty對於此已經有考慮，還有別的解碼器可以幫助我們解決問題，
到此這篇關於Netty解決 TCP 粘包拆包的方法的文章就介紹到這瞭,更多相關Netty解決 TCP 粘包拆包內容請搜索WalkonNet以前的文章或繼續瀏覽下面的相關文章希望大傢以後多多支持WalkonNet！
推薦閱讀：
- 如何在Netty中註解使用Service或者Mapper
- 在SpringBoot中,如何使用Netty實現遠程調用方法總結
- Netty結合Protobuf進行編解碼的方法
- netty 實現tomcat的示例代碼
- Springboot+Netty+Websocket實現消息推送實例