Lesen Sie asynchron Daten von NetworkStream mit einer großen Menge von Paketen

Question

In meiner Anwendung hat jedes Paket 2 Bytes Länge am Start. Nach einiger Zeit beginnt die Anwendung jedoch, eine Länge von weniger als Null zu empfangen. Im synchronen Client funktioniert alles korrekt, aber es ist zu langsam. Ich bin 100% sicher, dass im Server alles korrekt ist.

Connect:

public void Connect(IPAddress ip, int port) 
    { 
     tcpClient.Connect(ip, port); 
     stream = tcpClient.GetStream(); 
     byte[] len_buffer = new byte[2]; 
     stream.BeginRead(len_buffer, 0, len_buffer.Length, OnDataRead, len_buffer); 
    } 

OnDataRead:

private void OnDataRead(IAsyncResult ar) 
    { 
      byte[] len = ar.AsyncState as byte[]; 
      int length = BitConverter.ToInt16(len, 0); 
      byte[] buffer = new byte[length]; 

      int remaining = length; 
      int pos = 0; 
      while (remaining != 0) 
      { 
       int add = stream.Read(buffer, pos, remaining); 
       pos += add; 
       remaining -= add; 
      } 
      Process(buffer); 
      len = new byte[2]; 

      stream.EndRead(ar); 
      stream.BeginRead(len, 0, len.Length, OnDataRead, len); 
    } 

Answer 1

Wie ich sehe, Sie vermischen synchronious und asynchronious. Das ist eine schlechte Übung.

Was Sie wollen, ist so etwas wie:

var header = ReadHeader(); // 2 bytes 
var data = ReadData(header.DataSize); 

ich nicht den Netzwerk-Stream verwendet haben, aber .... Hier ist ein Beispiel meiner async SocketReader:

public static class SocketReader 
{ 
    // This method will continues read until count bytes are read. (or socket is closed) 
    private static void DoReadFromSocket(Socket socket, int bytesRead, int count, byte[] buffer, Action<ArraySegment<byte>> endRead) 
    { 
     // Start a BeginReceive. 
     try 
     { 
      socket.BeginReceive(buffer, bytesRead, count - bytesRead, SocketFlags.None, (asyncResult) => 
      { 
       // Get the bytes read. 
       int read = 0; 
       try 
       { 
        // if this goes wrong, the read remains 0 
        read = socket.EndReceive(asyncResult); 
       } 
       catch (ObjectDisposedException) { } 
       catch (Exception exception) 
       { 
        Trace.TraceError(exception.Message); 
       } 


       // if zero bytes received, the socket isn't available anymore. 
       if (read == 0) 
       { 
        endRead(new ArraySegment<byte>(buffer, 0, 0)); 
        return; 
       } 

       // increase the bytesRead, (position within the buffer) 
       bytesRead += read; 

       // if all bytes are read, call the endRead with the buffer. 
       if (bytesRead == count) 
        // All bytes are read. Invoke callback. 
        endRead(new ArraySegment<byte>(buffer, 0, count)); 
       else 
        // if not all bytes received, start another BeginReceive. 
        DoReadFromSocket(socket, bytesRead, count, buffer, endRead); 

      }, null); 
     } 
     catch (Exception exception) 
     { 
      Trace.TraceError(exception.Message); 
      endRead(new ArraySegment<byte>(buffer, 0, 0)); 
     } 
    } 

    public static void ReadFromSocket(Socket socket, int count, Action<ArraySegment<byte>> endRead) 
    { 
     // read from socket, construct a new buffer. 
     DoReadFromSocket(socket, 0, count, new byte[count], endRead); 
    } 

    public static void ReadFromSocket(Socket socket, int count, byte[] buffer, Action<ArraySegment<byte>> endRead) 
    { 
     // if you do have a buffer available, you can pass that one. (this way you do not construct new buffers for receiving and able to reuse buffers) 

     // if the buffer is too small, raise an exception, the caller should check the count and size of the buffer. 
     if (count > buffer.Length) 
      throw new ArgumentOutOfRangeException(nameof(count)); 

     DoReadFromSocket(socket, 0, count, buffer, endRead); 
    } 
} 

Verwendung:

SocketReader.ReadFromSocket(socket, 2, (headerData) => 
{ 
    if(headerData.Count == 0) 
    { 
     // nothing/closed 
     return; 
    } 

    // Read the length of the data. 
    int length = BitConverter.ToInt16(headerData.Array, headerData.Offset); 

    SocketReader.ReadFromSocket(socket, length, (dataBufferSegment) => 
    { 
     if(dataBufferSegment.Count == 0) 
     { 
      // nothing/closed 
      return; 
     } 

     Process(dataBufferSegment); 

     // extra: if you need a binaryreader.. 
     using(var stream = new MemoryStream(dataBufferSegment.Array, dataBufferSegment.Offset, dataBufferSegment.Count)) 
     using(var reader = new BinaryReader(stream)) 
     { 
      var whatever = reader.ReadInt32(); 
     } 
    } 
}); 

Sie können das optimieren Empfangspuffer durch einen Puffer übergeben (siehe den Überlastungen)

---

weiterhin empfangen:(Wiederverwendung receivebuffer)

public class PacketReader 
{ 
    private byte[] _receiveBuffer = new byte[2]; 

    // This will run until the socket is closed.  
    public void StartReceiving(Socket socket, Action<ArraySegment<byte>> process) 
    { 
     SocketReader.ReadFromSocket(socket, 2, _receiveBuffer, (headerData) => 
     { 
      if(headerData.Count == 0) 
      { 
       // nothing/closed 
       return; 
      } 

      // Read the length of the data. 
      int length = BitConverter.ToInt16(headerData.Array, headerData.Offset); 

      // if the receive buffer is too small, reallocate it. 
      if(_receiveBuffer.Length < length) 
       _receiveBuffer = new byte[length]; 

      SocketReader.ReadFromSocket(socket, length, _receiveBuffer, (dataBufferSegment) => 
      { 
       if(dataBufferSegment.Count == 0) 
       { 
        // nothing/closed 
        return; 
       } 

       try 
       { 
        process(dataBufferSegment); 
       } 
       catch { } 

       StartReceiving(socket, process); 
      }); 
     }); 
    } 
} 

Verbrauch:

private PacketReader _reader; 

public void Start() 
{ 
    _reader = new PacketReader(socket, HandlePacket); 
} 

private void HandlePacket(ArraySegment<byte> packet) 
{ 
    // do stuff..... 
}