1. 程式人生 > >java nio訊息半包、粘包解決方案

java nio訊息半包、粘包解決方案

問題背景

NIO是面向緩衝區進行通訊的,不是面向流的。我們都知道,既然是緩衝區,那它一定存在一個固定大小。這樣一來通常會遇到兩個問題:

  • 訊息粘包:當緩衝區足夠大,由於網路不穩定種種原因,可能會有多條訊息從通道讀入緩衝區,此時如果無法分清資料包之間的界限,就會導致粘包問題;
  • 訊息不完整:若訊息沒有接收完,緩衝區就被填滿了,會導致從緩衝區取出的訊息不完整,即半包的現象。

介紹這個問題之前,務必要提一下我程式碼整體架構。
程式碼參見GitHub倉庫

https://github.com/CuriousLei/smyl-im

在這個專案中,我的NIO核心庫設計思路流程圖如下所示

介紹:

  • 服務端為每一個連線上的客戶端建立一個Connector物件,為其提供IO服務;
  • ioArgs物件內部例項域引用了緩衝區buffer,作為直接與channel進行資料互動的緩衝區;
  • 兩個執行緒池,分別操控ioArgs進行讀和寫操作;
  • connector與ioArgs關係:(1)輸入,執行緒池處理讀事件,資料寫入ioArgs,並回調給connector;(2)輸出,connector將資料寫入ioArgs,將ioArgs傳入Runnable物件,供執行緒池處理;
  • 兩個selector執行緒,分別監聽channel的讀和寫事件。事件就緒,則觸發執行緒池工作。

思路

光這樣實現,必然會有粘包、半包問題。要重現這兩個問題也很簡單。

  • ioArgs中把緩衝區設定小一點,傳送一條大於該長度的資料,服務端會當成兩條訊息讀取,即訊息不完整;
  • 線上程程式碼中,加一個Thread.sleep()延時等待,客戶端連續發幾條訊息(總長度小於緩衝區大小),也可以重現粘包現象。

這個問題實質上是訊息體與緩衝區資料不一一對應導致的。那麼,如何解決呢?

固定頭部方案

可以採用固定頭部方案來解決,頭部設定四個位元組,儲存一個int值,記錄後面資料的長度。以此來標記一個訊息體。

  • 讀取資料時,根據頭部的長度資訊,按序讀取ioArgs緩衝區中的資料,若沒有達到長度要求,繼續讀下一個ioArgs。這樣自然不會出現粘包、半包問題。
  • 輸出資料時,也採用同樣的機制封裝資料,首部四個位元組記錄長度。

我的工程專案中,客戶端和服務端共用一個nio核心包,即niohdl,可保證收發資料格式一致。

設計方案

要實現以上設想,必須在connector和ioArgs之間加一層Dispatcher類,用於處理訊息體與緩衝區之間的轉化關係(訊息體取個名字:Packet)。根據輸入和輸出的不同,分別叫ReceiveDispatcher和SendDispatcher。即通過它們來操作Packet與ioArgs之間的轉化。

Packet

定義這個訊息體,繼承關係如下圖所示:

Packet是基類,程式碼如下:

package cn.buptleida.niohdl.core;
import java.io.Closeable;
import java.io.IOException;
/**
 * 公共的資料封裝
 * 提供了型別以及基本的長度的定義
 */
public class Packet implements Closeable {
    protected byte type;
    protected int length;

    public byte type(){
        return type;
    }

    public int length(){
        return length;
    }

    @Override
    public void close() throws IOException {

    }
}

SendPacket和ReceivePacket分別代表傳送訊息體和接收訊息體。StringReceivePacket和StringSendPacket代表字串類的訊息,因為本次實踐只限於字串訊息的收發,今後可能有檔案之類的,有待擴充套件。

程式碼中必然會涉及到位元組陣列的操作,所以,以StringSendPacket為例,需要提供將String轉化為byte[]的方法。程式碼如下所示:

package cn.buptleida.niohdl.box;
import cn.buptleida.niohdl.core.SendPacket;

public class StringSendPacket extends SendPacket {

    private final byte[] bytes;

    public StringSendPacket(String msg) {
        this.bytes = msg.getBytes();
        this.length = bytes.length;//父類中的例項域
    }

    @Override
    public byte[] bytes() {
        return bytes;
    }
}

SendDispatcher

在connector物件的例項域中會引用一個SendDispatcher物件。傳送資料時,會通過SendDispatcher中的方法對資料進行封裝和處理。其大致的關係圖如下所示:

SendDispatcher中設定任務佇列Queue

在程式碼中,SendDispatcher實際上是一個介面,我用AsyncSendDispatcher實現此介面,程式碼如下:

package cn.buptleida.niohdl.impl.async;

import cn.buptleida.niohdl.core.*;
import cn.buptleida.utils.CloseUtil;

import java.io.IOException;
import java.util.Queue;
import java.util.concurrent.ConcurrentLinkedDeque;
import java.util.concurrent.atomic.AtomicBoolean;

public class AsyncSendDispatcher implements SendDispatcher {
    private final AtomicBoolean isClosed = new AtomicBoolean(false);
    private Sender sender;
    private Queue<SendPacket> queue = new ConcurrentLinkedDeque<>();
    private AtomicBoolean isSending = new AtomicBoolean();
    private ioArgs ioArgs = new ioArgs();
    private SendPacket packetTemp;
    //當前傳送的packet大小以及進度
    private int total;
    private int position;

    public AsyncSendDispatcher(Sender sender) {
        this.sender = sender;
    }

    /**
     * connector將資料封裝進packet後,呼叫這個方法
     * @param packet
     */
    @Override
    public void send(SendPacket packet) {
        queue.offer(packet);//將資料放進佇列中
        if (isSending.compareAndSet(false, true)) {
            sendNextPacket();
        }
    }
    
    @Override
    public void cancel(SendPacket packet) {

    }

    /**
     * 從佇列中取資料
     * @return
     */
    private SendPacket takePacket() {
        SendPacket packet = queue.poll();
        if (packet != null && packet.isCanceled()) {
            //已經取消不用傳送
            return takePacket();
        }
        return packet;
    }

    private void sendNextPacket() {
        SendPacket temp = packetTemp;
        if (temp != null) {
            CloseUtil.close(temp);
        }
        SendPacket packet = packetTemp = takePacket();
        if (packet == null) {
            //佇列為空,取消傳送狀態
            isSending.set(false);
            return;
        }

        total = packet.length();
        position = 0;

        sendCurrentPacket();
    }

    private void sendCurrentPacket() {
        ioArgs args = ioArgs;

        args.startWriting();//將ioArgs緩衝區中的指標設定好

        if (position >= total) {
            sendNextPacket();
            return;
        } else if (position == 0) {
            //首包,需要攜帶長度資訊
            args.writeLength(total);
        }

        byte[] bytes = packetTemp.bytes();
        //把bytes的資料寫入到IoArgs中
        int count = args.readFrom(bytes, position);
        position += count;

        //完成封裝
        args.finishWriting();//flip()操作
        //向通道註冊OP_write,將Args附加到runnable中;selector執行緒監聽到就緒即可觸發執行緒池進行訊息傳送
        try {
            sender.sendAsync(args, ioArgsEventListener);
        } catch (IOException e) {
            closeAndNotify();
        }
    }

    private void closeAndNotify() {
        CloseUtil.close(this);
    }

    @Override
    public void close(){
        if (isClosed.compareAndSet(false, true)) {
            isSending.set(false);
            SendPacket packet = packetTemp;
            if (packet != null) {
                packetTemp = null;
                CloseUtil.close(packet);
            }
        }
    }

    /**
     * 接收回調,來自writeHandler輸出執行緒
     */
    private ioArgs.IoArgsEventListener ioArgsEventListener = new ioArgs.IoArgsEventListener() {
        @Override
        public void onStarted(ioArgs args) {

        }

        @Override
        public void onCompleted(ioArgs args) {
            //繼續傳送當前包packetTemp,因為可能一個包沒發完
            sendCurrentPacket();
        }
    };


}

ReceiveDispatcher

同樣,ReceiveDispatcher也是一個介面,程式碼中用AsyncReceiveDispatcher實現。在connector物件的例項域中會引用一個AsyncReceiveDispatcher物件。接收資料時,會通過ReceiveDispatcher中的方法對接收到的資料進行拆包處理。其大致的關係圖如下所示:

每一個訊息體的首部會有一個四位元組的int欄位,代表訊息的長度值,按照這個長度來進行讀取。如若一個ioArgs未滿足這個長度,就讀取下一個ioArgs,保證資料包的完整性。這個流程就不畫程式框圖了,偷個懶hhhh。其實看下面程式碼註釋已經很清晰了,容易理解。

AsyncReceiveDispatcher的程式碼如下所示:

package cn.buptleida.niohdl.impl.async;

import cn.buptleida.niohdl.box.StringReceivePacket;
import cn.buptleida.niohdl.core.ReceiveDispatcher;
import cn.buptleida.niohdl.core.ReceivePacket;
import cn.buptleida.niohdl.core.Receiver;
import cn.buptleida.niohdl.core.ioArgs;
import cn.buptleida.utils.CloseUtil;

import java.io.IOException;
import java.util.concurrent.atomic.AtomicBoolean;

public class AsyncReceiveDispatcher implements ReceiveDispatcher {
    private final AtomicBoolean isClosed = new AtomicBoolean(false);
    private final Receiver receiver;
    private final ReceivePacketCallback callback;
    private ioArgs args = new ioArgs();
    private ReceivePacket packetTemp;
    private byte[] buffer;
    private int total;
    private int position;

    public AsyncReceiveDispatcher(Receiver receiver, ReceivePacketCallback callback) {
        this.receiver = receiver;
        this.receiver.setReceiveListener(ioArgsEventListener);
        this.callback = callback;
    }

    /**
     * connector中呼叫該方法進行
     */
    @Override
    public void start() {
        registerReceive();
    }

    private void registerReceive() {

        try {
            receiver.receiveAsync(args);
        } catch (IOException e) {
            closeAndNotify();
        }
    }

    private void closeAndNotify() {
        CloseUtil.close(this);
    }

    @Override
    public void stop() {

    }

    @Override
    public void close() throws IOException {
        if(isClosed.compareAndSet(false,true)){
            ReceivePacket packet = packetTemp;
            if(packet!=null){
                packetTemp = null;
                CloseUtil.close(packet);
            }
        }
    }

    /**
     * 回撥方法,從readHandler輸入執行緒中回撥
     */
    private ioArgs.IoArgsEventListener ioArgsEventListener = new ioArgs.IoArgsEventListener() {
        @Override
        public void onStarted(ioArgs args) {
            int receiveSize;
            if (packetTemp == null) {
                receiveSize = 4;
            } else {
                receiveSize = Math.min(total - position, args.capacity());
            }
            //設定接受資料大小
            args.setLimit(receiveSize);
        }

        @Override
        public void onCompleted(ioArgs args) {
            assemblePacket(args);
            //繼續接受下一條資料,因為可能同一個訊息可能分隔在兩份IoArgs中
            registerReceive();
        }
    };

    /**
     * 解析資料到packet
     * @param args
     */
    private void assemblePacket(ioArgs args) {
        if (packetTemp == null) {
            int length = args.readLength();
            packetTemp = new StringReceivePacket(length);
            buffer = new byte[length];
            total = length;
            position = 0;
        }
        //將args中的資料寫進外面buffer中
        int count = args.writeTo(buffer,0);
        if(count>0){
            //將資料存進StringReceivePacket的buffer當中
            packetTemp.save(buffer,count);
            position+=count;
            
            if(position == total){
                completePacket();
                packetTemp = null;
            }
        }
        
    }

    private void completePacket() {
        ReceivePacket packet = this.packetTemp;
        CloseUtil.close(packet);
        callback.onReceivePacketCompleted(packet);
    }

}

總結

其實粘包、半包的解決方案並沒有什麼奧祕,單純地複雜而已。方法核心就是自定義一個訊息體Packet,完成Packet中的byte陣列與緩衝區陣列之間的複製轉化即可。當然,position、limit等等指標的輔助很重要。

總結這個部落格,也是將目前為止的工作進行梳理和記錄。我將通過smyl-im這個專案來持續學習+實踐。因為之前學習過程中有很多零碎的知識點,都躺在我的有道雲筆記裡,感覺沒必要總結成部落格。本次部落格講的內容剛好是一個成體系的東西,正好可以將這個專案背景帶出來,後續的部落格就可以在這基礎上衍生拓展了