C#處理TCP數(shù)據(jù)的方法詳解

更新時間：2024年06月12日 09:14:47 作者：碼農浩克

Tcp是一個面向連接的流數(shù)據(jù)傳輸協(xié)議,用人話說就是傳輸是一個已經(jīng)建立好連接的管道,數(shù)據(jù)都在管道里像流水一樣流淌到對端,那么數(shù)據(jù)必然存在幾個問題,比如數(shù)據(jù)如何持續(xù)的讀取,數(shù)據(jù)包的邊界等,本文給大家介紹了C#處理TCP數(shù)據(jù)的方法,需要的朋友可以參考下

前言

Tcp是一個面向連接的流數(shù)據(jù)傳輸協(xié)議，用人話說就是傳輸是一個已經(jīng)建立好連接的管道，數(shù)據(jù)都在管道里像流水一樣流淌到對端。

那么數(shù)據(jù)必然存在幾個問題，比如數(shù)據(jù)如何持續(xù)的讀取，數(shù)據(jù)包的邊界等。

Nagle's算法

Nagle 算法的核心思想是，在一個 TCP 連接上，最多只能有一個未被確認的小數(shù)據(jù)包（小于 MSS，即最大報文段大?。?/p>

優(yōu)勢

減少網(wǎng)絡擁塞：通過合并小數(shù)據(jù)包，減少了網(wǎng)絡中的數(shù)據(jù)包數(shù)量，降低了擁塞的可能性。

提高網(wǎng)絡效率：在低速網(wǎng)絡中，Nagle 算法可以顯著提高傳輸效率。

劣勢

增加延遲：在交互式應用中，Nagle 算法可能導致顯著的延遲，因為它等待 ACK 或合并數(shù)據(jù)包。

C#中如何配置?

var _socket = new Socket(IPAddress.Any.AddressFamily, SocketType.Stream, ProtocolType.Tcp);
_serverSocket.NoDelay = _options.NoDelay;

連接超時

在調用客戶端Socket連接服務器的時候，可以設置連接超時機制，具體可以傳入一個任務的取消令牌，并且設置超時時間。

CancellationTokenSource connectTokenSource = new CancellationTokenSource();
connectTokenSource.CancelAfter(3000); //3秒
await _socket.ConnectAsync(RemoteEndPoint, connectTokenSource.Token);

SSL加密傳輸

TCP使用SSL加密傳輸，通過非對稱加密的方式，利用證書，保證雙方使用了安全的密鑰加密了報文。在C#中如何配置?

服務端配置

//創(chuàng)建證書對象
var _certificate  = _certificate = new X509Certificate2(_options.PfxCertFilename, _options.PfxPassword);
 
//與客戶端進行驗證
if (allowingUntrustedSSLCertificate) //是否允許不受信任的證書
{
    SslStream = new SslStream(NetworkStream, false,
        (obj, certificate, chain, error) => true);
}
else
{
    SslStream = new SslStream(NetworkStream, false);
}
 
try
{
    //serverCertificate：用于對服務器進行身份驗證的 X509Certificate
    //clientCertificateRequired：一個 Boolean 值，指定客戶端是否必須為身份驗證提供證書
    //checkCertificateRevocation：一個 Boolean 值，指定在身份驗證過程中是否檢查證書吊銷列表
    await SslStream.AuthenticateAsServerAsync(new SslServerAuthenticationOptions()
    {
        ServerCertificate = x509Certificate,
        ClientCertificateRequired = mutuallyAuthenticate,
        CertificateRevocationCheckMode = checkCertificateRevocation ? X509RevocationMode.Online : X509RevocationMode.NoCheck
    }, cancellationToken).ConfigureAwait(false);
 
    if (!SslStream.IsEncrypted || !SslStream.IsAuthenticated)
    {
        return false;
    }
 
    if (mutuallyAuthenticate && !SslStream.IsMutuallyAuthenticated)
    {
        return false;
    }
}
catch (Exception)
{
    throw;
}
 
//完成驗證后，通過SslStream傳輸數(shù)據(jù)
int readCount = await SslStream.ReadAsync(buffer, _lifecycleTokenSource.Token)
    .ConfigureAwait(false);

客戶端配置

var _certificate = new X509Certificate2(_options.PfxCertFilename, _options.PfxPassword);
 
if (_options.IsSsl) //如果使用ssl加密傳輸
{
    if (_options.AllowingUntrustedSSLCertificate)//是否允許不受信任的證書
    {
        _sslStream = new SslStream(_networkStream, false,
                (obj, certificate, chain, error) => true);
    }
    else
    {
        _sslStream = new SslStream(_networkStream, false);
    }
 
    _sslStream.ReadTimeout = _options.ReadTimeout;
    _sslStream.WriteTimeout = _options.WriteTimeout;
    await _sslStream.AuthenticateAsClientAsync(new SslClientAuthenticationOptions()
    {
        TargetHost = RemoteEndPoint.Address.ToString(),
        EnabledSslProtocols = System.Security.Authentication.SslProtocols.Tls12,
        CertificateRevocationCheckMode = _options.CheckCertificateRevocation ? X509RevocationMode.Online : X509RevocationMode.NoCheck,
        ClientCertificates = new X509CertificateCollection() { _certificate }
    }, connectTokenSource.Token).ConfigureAwait(false);
 
    if (!_sslStream.IsEncrypted || !_sslStream.IsAuthenticated ||
        (_options.MutuallyAuthenticate && !_sslStream.IsMutuallyAuthenticated))
    {
        throw new InvalidOperationException("SSL authenticated faild!");
    }
}

KeepAlive

keepAlive不是TCP協(xié)議中的，而是各個操作系統(tǒng)本身實現(xiàn)的功能，主要是防止一些Socket突然斷開后沒有被感知到，導致一直浪費資源的情況。

/// <summary>
/// 開啟Socket的KeepAlive
/// 設置tcp協(xié)議的一些KeepAlive參數(shù)
/// </summary>
/// <param name="socket"></param>
/// <param name="tcpKeepAliveInterval">沒有接收到對方確認，繼續(xù)發(fā)送KeepAlive的發(fā)送頻率</param>
/// <param name="tcpKeepAliveTime">KeepAlive的空閑時長，或者說每次正常發(fā)送心跳的周期</param>
/// <param name="tcpKeepAliveRetryCount">KeepAlive之后設置最大允許發(fā)送?；钐綔y包的次數(shù)，到達此次數(shù)后直接放棄嘗試，并關閉連接</param>
internal static void SetKeepAlive(this Socket socket, int tcpKeepAliveInterval, int tcpKeepAliveTime, int tcpKeepAliveRetryCount)
{
    socket.SetSocketOption(SocketOptionLevel.Socket, SocketOptionName.KeepAlive, true);
    socket.SetSocketOption(SocketOptionLevel.Tcp, SocketOptionName.TcpKeepAliveInterval, tcpKeepAliveInterval);
    socket.SetSocketOption(SocketOptionLevel.Tcp, SocketOptionName.TcpKeepAliveTime, tcpKeepAliveTime);
    socket.SetSocketOption(SocketOptionLevel.Tcp, SocketOptionName.TcpKeepAliveRetryCount, tcpKeepAliveRetryCount);
}

具體的開啟，還需要看操作系統(tǒng)的版本以及不同操作系統(tǒng)的支持。

粘包斷包處理

Pipe & ReadOnlySequence

TCP面向應用是流式數(shù)據(jù)傳輸，所以接收端接到的數(shù)據(jù)是像流水一樣從管道中傳來，每次取到的數(shù)據(jù)取決于應用設置的緩沖區(qū)大小，以及套接字本身緩沖區(qū)待讀取字節(jié)數(shù)。

C#中提供的Pipe就如上圖一樣，是一個管道Pipe有兩個對象成員，一個是PipeWriter,一個是PipeReader,可以理解為一個是生產者，專門往管道里灌輸數(shù)據(jù)流，即字節(jié)流，一個是消費者，專門從管道里獲取字節(jié)流進行處理。

可以看到Pipe中的數(shù)據(jù)包是用鏈表關聯(lián)的，但是這個數(shù)據(jù)包是從Socke緩沖區(qū)每次取到的數(shù)據(jù)包，它不一定是一個完整的數(shù)據(jù)包，所以這些數(shù)據(jù)包連接起來后形成了一個C#提供的另外一個抽象的對象ReadOnlySequence。

但是這里還是沒有提供太好的處理斷包和粘包的辦法，因為斷包粘包的處理需要兩方面

1、業(yè)務數(shù)據(jù)包的定義

2、數(shù)據(jù)流切割出一個個完整的數(shù)據(jù)包

假設業(yè)務已經(jīng)定義好了數(shù)據(jù)包，那么我們如何從Pipe中這些數(shù)據(jù)包根據(jù)業(yè)務定義來從不同的數(shù)據(jù)包中切割出一個完整的包，那么就需要ReadOnlySequence,它提供的操作方法，非常方便我們去切割數(shù)據(jù)，主要是頭尾數(shù)據(jù)包的切割。

假設我們業(yè)務層定義了一個數(shù)據(jù)包結構，數(shù)據(jù)包是不定長的，包體長度每次都寫在包頭里，我們來實現(xiàn)一個數(shù)據(jù)包過濾器。

//收到消息
 while (!_receiveDataTokenSource.Token.IsCancellationRequested)
 {
     try
     {
        //從pipe中獲取緩沖區(qū)
         Memory<byte> buffer = _pipeWriter.GetMemory(_options.BufferSize);
         int readCount = 0;
         readCount = await _sslStream.ReadAsync(buffer, _lifecycleTokenSource.Token).ConfigureAwait(false);
 
         if (readCount > 0)
         {
 
             var data = buffer.Slice(0, readCount);
             //告知消費者，往Pipe的管道中寫入了多少字節(jié)數(shù)據(jù)
             _pipeWriter.Advance(readCount);
         }
         else
         {
             if (IsDisconnect())
             {
                 await DisConnectAsync();
             }
 
             throw new SocketException();
         }
 
         FlushResult result = await _pipeWriter.FlushAsync().ConfigureAwait(false);
         if (result.IsCompleted)
         {
             break;
         }
     }
     catch (IOException)
     {
         //TODO log
         break;
     }
     catch (SocketException)
     {
         //TODO log
         break;
     }
     catch (TaskCanceledException)
     {
         //TODO log
         break;
     }
 }
 
 _pipeWriter.Complete();
//消費者處理數(shù)據(jù)
 while (!_lifecycleTokenSource.Token.IsCancellationRequested)
 {
     ReadResult result = await _pipeReader.ReadAsync();
     ReadOnlySequence<byte> buffer = result.Buffer;
     ReadOnlySequence<byte> data;
     do
     {
        //通過過濾器得到一個完整的包
         data = _receivePackageFilter.ResolvePackage(ref buffer);
 
         if (!data.IsEmpty)
         {
             OnReceivedData?.Invoke(this, new ClientDataReceiveEventArgs(data.ToArray()));
         }
     }
     while (!data.IsEmpty && buffer.Length > 0);
     _pipeReader.AdvanceTo(buffer.Start);
 }
 
 _pipeReader.Complete();
/// <summary>
/// 解析數(shù)據(jù)包
/// 固定報文頭解析協(xié)議
/// </summary>
/// <param name="headerSize">數(shù)據(jù)報文頭的大小</param>
/// <param name="bodyLengthIndex">數(shù)據(jù)包大小在報文頭中的位置</param>
/// <param name="bodyLengthBytes">數(shù)據(jù)包大小在報文頭中的長度</param>
/// <param name="IsLittleEndian">數(shù)據(jù)報文大小端。windows中通常是小端，unix通常是大端模式</param>
/// </summary>
/// <param name="sequence">一個完整的業(yè)務數(shù)據(jù)包</param>
public override ReadOnlySequence<byte> ResolvePackage(ref ReadOnlySequence<byte> sequence)
{
    var len = sequence.Length;
    if (len < _bodyLengthIndex) return default;
    var bodyLengthSequence = sequence.Slice(_bodyLengthIndex, _bodyLengthBytes);
    byte[] bodyLengthBytes = ArrayPool<byte>.Shared.Rent(_bodyLengthBytes);
    try
    {
        int index = 0;
        foreach (var item in bodyLengthSequence)
        {
            Array.Copy(item.ToArray(), 0, bodyLengthBytes, index, item.Length);
            index += item.Length;
        }
 
        long bodyLength = 0;
        int offset = 0;
        if (!_isLittleEndian)
        {
            offset = bodyLengthBytes.Length - 1;
            foreach (var bytes in bodyLengthBytes)
            {
                bodyLength += bytes << (offset * 8);
                offset--;
            }
        }
        else
        {
 
            foreach (var bytes in bodyLengthBytes)
            {
                bodyLength += bytes << (offset * 8);
                offset++;
            }
        }
 
        if (sequence.Length < _headerSize + bodyLength)
            return default;
 
        var endPosition = sequence.GetPosition(_headerSize + bodyLength);
        var data = sequence.Slice(0, endPosition);//得到完整數(shù)據(jù)包
        sequence = sequence.Slice(endPosition);//緩沖區(qū)中去除取到的完整包
 
        return data;
    }
    finally
    {
        ArrayPool<byte>.Shared.Return(bodyLengthBytes);
    }
}

以上就是實現(xiàn)了固定數(shù)據(jù)包頭實現(xiàn)粘包斷包處理的部分代碼。

關于TCP的連接還有一些，比如客戶端連接限制，空閑連接關閉等。

到此這篇關于C#處理TCP數(shù)據(jù)的方法詳解的文章就介紹到這了,更多相關C#處理TCP數(shù)據(jù)內容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關文章希望大家以后多多支持腳本之家！

您可能感興趣的文章: