Netty 是如何解決 TCP 粘包拆包的?
我們都知道TCP是基于字節(jié)流的傳輸協(xié)議。
那么數(shù)據(jù)在通信層傳播其實(shí)就像河水一樣并沒(méi)有明顯的分界線,而數(shù)據(jù)具體表示什么意思什么地方有句號(hào)什么地方有分號(hào)這個(gè)對(duì)于TCP底層來(lái)說(shuō)并不清楚。應(yīng)用層向TCP層發(fā)送用于網(wǎng)間傳輸?shù)?、?位字節(jié)表示的數(shù)據(jù)流,然后TCP把數(shù)據(jù)流分區(qū)成適當(dāng)長(zhǎng)度的報(bào)文段,之后TCP把結(jié)果包傳給IP層,由它來(lái)通過(guò)網(wǎng)絡(luò)將包傳送給接收端實(shí)體的TCP層。
所以對(duì)于這個(gè)數(shù)據(jù)拆分成大包小包的問(wèn)題就是我們今天要講的粘包和拆包的問(wèn)題。
1、TCP粘包拆包問(wèn)題說(shuō)明
粘包和拆包這兩個(gè)概念估計(jì)大家還不清楚,通過(guò)下面這張圖我們來(lái)分析一下:
假設(shè)客戶端分別發(fā)送兩個(gè)數(shù)據(jù)包D1,D2個(gè)服務(wù)端,但是發(fā)送過(guò)程中數(shù)據(jù)是何種形式進(jìn)行傳播這個(gè)并不清楚,分別有下列4種情況:
- 服務(wù)端一次接受到了D1和D2兩個(gè)數(shù)據(jù)包,兩個(gè)包粘在一起,稱為粘包;
- 服務(wù)端分兩次讀取到數(shù)據(jù)包D1和D2,沒(méi)有發(fā)生粘包和拆包;
- 服務(wù)端分兩次讀到了數(shù)據(jù)包,第一次讀到了D1和D2的部分內(nèi)容,第二次讀到了D2的剩下部分,這個(gè)稱為拆包;
- 服務(wù)器分三次讀到了數(shù)據(jù)部分,第一次讀到了D1包,第二次讀到了D2包的部分內(nèi)容,第三次讀到了D2包的剩下內(nèi)容。
2、TCP粘包產(chǎn)生原因
我們知道在TCP協(xié)議中,應(yīng)用數(shù)據(jù)分割成TCP認(rèn)為最適合發(fā)送的數(shù)據(jù)塊,這部分是通過(guò)“MSS”(最大數(shù)據(jù)包長(zhǎng)度)選項(xiàng)來(lái)控制的,通常這種機(jī)制也被稱為一種協(xié)商機(jī)制,MSS規(guī)定了TCP傳往另一端的最大數(shù)據(jù)塊的長(zhǎng)度。
這個(gè)值TCP協(xié)議在實(shí)現(xiàn)的時(shí)候往往用MTU值代替(需要減去IP數(shù)據(jù)包包頭的大小20Bytes和TCP數(shù)據(jù)段的包頭20Bytes)所以往往MSS為1460。通訊雙方會(huì)根據(jù)雙方提供的MSS值得最小值確定為這次連接的最大MSS值。
tcp為提高性能,發(fā)送端會(huì)將需要發(fā)送的數(shù)據(jù)發(fā)送到緩沖區(qū),等待緩沖區(qū)滿了之后,再將緩沖中的數(shù)據(jù)發(fā)送到接收方。同理,接收方也有緩沖區(qū)這樣的機(jī)制,來(lái)接收數(shù)據(jù)。
發(fā)生粘包拆包的原因主要有以下這些:
- 應(yīng)用程序?qū)懭霐?shù)據(jù)的字節(jié)大小大于套接字發(fā)送緩沖區(qū)的大小將發(fā)生拆包;
- 進(jìn)行MSS大小的TCP分段。MSS是TCP報(bào)文段中的數(shù)據(jù)字段的最大長(zhǎng)度,當(dāng)TCP報(bào)文長(zhǎng)度-TCP頭部長(zhǎng)度>mss的時(shí)候?qū)l(fā)生拆包;
- 應(yīng)用程序?qū)懭霐?shù)據(jù)小于套接字緩沖區(qū)大小,網(wǎng)卡將應(yīng)用多次寫(xiě)入的數(shù)據(jù)發(fā)送到網(wǎng)絡(luò)上,將發(fā)生粘包;
- 數(shù)據(jù)包大于MTU的時(shí)候?qū)?huì)進(jìn)行切片。MTU即(Maxitum Transmission Unit) 最大傳輸單元,由于以太網(wǎng)傳輸電氣方面的限制,每個(gè)以太網(wǎng)幀都有最小的大小64bytes最大不能超過(guò)1518bytes,刨去以太網(wǎng)幀的幀頭14Bytes和幀尾CRC校驗(yàn)部分4Bytes,那么剩下承載上層協(xié)議的地方也就是Data域最大就只能有1500Bytes這個(gè)值我們就把它稱之為MTU。這個(gè)就是網(wǎng)絡(luò)層協(xié)議非常關(guān)心的地方,因?yàn)榫W(wǎng)絡(luò)層協(xié)議比如IP協(xié)議會(huì)根據(jù)這個(gè)值來(lái)決定是否把上層傳下來(lái)的數(shù)據(jù)進(jìn)行分片。
3、如何解決TCP粘包拆包
我們知道tcp是無(wú)界的數(shù)據(jù)流,且協(xié)議本身無(wú)法避免粘包,拆包的發(fā)生,那我們只能在應(yīng)用層數(shù)據(jù)協(xié)議上,加以控制。通常在制定傳輸數(shù)據(jù)時(shí),可以使用如下方法:
- 設(shè)置定長(zhǎng)消息,服務(wù)端每次讀取既定長(zhǎng)度的內(nèi)容作為一條完整消息;
- 使用帶消息頭的協(xié)議、消息頭存儲(chǔ)消息開(kāi)始標(biāo)識(shí)及消息長(zhǎng)度信息,服務(wù)端獲取消息頭的時(shí)候解析出消息長(zhǎng)度,然后向后讀取該長(zhǎng)度的內(nèi)容;
- 設(shè)置消息邊界,服務(wù)端從網(wǎng)絡(luò)流中按消息邊界分離出消息內(nèi)容。比如在消息末尾加上換行符用以區(qū)分消息結(jié)束。
當(dāng)然應(yīng)用層還有更多復(fù)雜的方式可以解決這個(gè)問(wèn)題,這個(gè)就屬于網(wǎng)絡(luò)層的問(wèn)題了,我們還是用java提供的方式來(lái)解決這個(gè)問(wèn)題。Spring Boot 學(xué)習(xí)筆記分享給你,我們先看一個(gè)例子看看粘包是如何發(fā)生的。
服務(wù)端:
- 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();
- }
- }
服務(wù)端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());
- }
- }
服務(wù)端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;
- //將上面的所有字符串作為一個(gè)消息體發(fā)送出去
- 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();
- }
- }
運(yùn)行客戶端和服務(wù)端我們能看到:
我們看到這個(gè)長(zhǎng)長(zhǎng)的字符串被截成了2段發(fā)送,這就是發(fā)生了拆包的現(xiàn)象。同樣粘包我們也很容易去模擬,我們把BaseClientHandler中的channelActive方法里面的:
- message = Unpooled.buffer(req.length);
- message.writeBytes(req);
- ctx.writeAndFlush(message);
這幾行代碼是把我們上面的一長(zhǎng)串字符轉(zhuǎn)成的byte數(shù)組寫(xiě)進(jìn)流里發(fā)送出去,那么我們可以在這里把上面發(fā)送消息的這幾行循環(huán)幾遍這樣發(fā)送的內(nèi)容增多了就有可能在拆包的時(shí)候把上一條消息的一部分分配到下一條消息里面了,修改如下:
- for (int i = 0; i < 3; i++) {
- message = Unpooled.buffer(req.length);
- message.writeBytes(req);
- ctx.writeAndFlush(message);
- }
改完之后我們?cè)龠\(yùn)行一下,輸出太長(zhǎng)不好截圖,我們?cè)谳敵鼋Y(jié)果中能看到循環(huán)3次之后的消息服務(wù)端收到的就不是之前的完整的一條了,而是被拆分了4次發(fā)送。
對(duì)于上面出現(xiàn)的粘包和拆包的問(wèn)題,Netty已有考慮,并且有實(shí)施的方案:LineBasedFrameDecoder。另外,微信搜索Java技術(shù)棧,在后臺(tái)回復(fù):面試,可以獲取我整理的 Java 系列面試題和答案。
我們重新改寫(xiě)一下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))。同時(shí),我們還得對(duì)上面發(fā)送的消息進(jìn)行改造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”,只保留字符串末尾的這一個(gè)。原因稍后再說(shuō)。channelActive方法中我們不必再用循環(huán)多次發(fā)送消息了,只發(fā)送一次就好(第一個(gè)例子中發(fā)送一次的時(shí)候是發(fā)生了拆包的),然后我們?cè)俅芜\(yùn)行,大家會(huì)看到這么長(zhǎng)一串字符只發(fā)送了一串就發(fā)送完畢。程序輸出我就不截圖了。下面來(lái)解釋一下LineBasedFrameDecoder。
LineBasedFrameDecoder的工作原理是它依次遍歷ByteBuf 中的可讀字節(jié),判斷看是否有”\n” 或者” \r\n”,如果有,就以此位置為結(jié)束位置,從可讀索引到結(jié)束位置區(qū)間的字節(jié)就組成了一行。它是以換行符為結(jié)束標(biāo)志的解碼器。支持?jǐn)y帶結(jié)束符或者不攜帶結(jié)束符兩種解碼方式,同時(shí)支持配置單行的最大長(zhǎng)度。如果連續(xù)讀取到最大長(zhǎng)度后仍然沒(méi)有發(fā)現(xiàn)換行符,就會(huì)拋出異常,同時(shí)忽略掉之前讀到的異常碼流。這個(gè)對(duì)于我們確定消息最大長(zhǎng)度的應(yīng)用場(chǎng)景還是很有幫助。
對(duì)于上面的判斷看是否有”\n” 或者” \r\n”以此作為結(jié)束的標(biāo)志我們可能回想,要是沒(méi)有”\n” 或者” \r\n”那還有什么別的方式可以判斷消息是否結(jié)束呢。別擔(dān)心,Netty對(duì)于此已經(jīng)有考慮,還有別的解碼器可以幫助我們解決問(wèn)題,另外,關(guān)注公眾號(hào)Java技術(shù)棧,在后臺(tái)回復(fù):面試,可以獲取我整理的 Java 系列面試題和答案,非常齊全。