使用 TcpClient/Socket 安全传输/发送大数据包

Question

我正在编写一个需要发送和接收数据的基于 TCP 的客户端。我使用了 .NET Framework 为 Socket 类提供的Asynchronous Programming Model (APM)。

连接到套接字后，我开始使用BeginReceive 等待套接字上的数据。

现在，当我等待 Socket 上的数据时，我可能需要通过套接字发送数据。而且send方法可以多次调用，

所以我已经确定了

• 之前Send 调用的所有字节都已完全发送。
• 我发送数据的方式是安全的，因为在数据发送过程中，可以进行任何发送数据的调用。

这是我在套接字上的第一个工作，那么我发送数据的方法是否正确？

    private readonly object writeLock = new object();
    public void Send(NetworkCommand cmd)
    {
        var data = cmd.ToBytesWithLengthPrefix();
        ThreadPool.QueueUserWorkItem(AsyncDataSent, data);
    }

    private int bytesSent;
    private void AsyncDataSent(object odata)
    {
        lock (writeLock)
        {
            var data = (byte[])odata;
            int total = data.Length;
            bytesSent = 0;
            int buf = Globals.BUFFER_SIZE;
            while (bytesSent < total)
            {
                if (total - bytesSent < Globals.BUFFER_SIZE)
                {
                    buf = total - bytesSent;
                }
                IAsyncResult ar = socket.BeginSend(data, bytesSent, buf, SocketFlags.None, DataSentCallback, data);
                ar.AsyncWaitHandle.WaitOne();
            }
        }
    }

对象如何变成byte[]，有时NetworkCommand可以大到0.5 MB

    public byte[] ToBytesWithLengthPrefix()
    {
        var stream = new MemoryStream();
        try
        {
            Serializer.SerializeWithLengthPrefix(stream, this, PrefixStyle.Fixed32);
            return stream.ToArray();
        }
        finally
        {
            stream.Close();
            stream.Dispose();
        }
    }

---

完整课程

namespace Cybotech.Network
{
    public delegate void ConnectedDelegate(IPEndPoint ep);
    public delegate void DisconnectedDelegate(IPEndPoint ep);
    public delegate void CommandReceivedDelagate(IPEndPoint ep, NetworkCommand cmd);
}


using System;
using System.Net;
using System.Net.Sockets;
using Cybotech.Helper;
using Cybotech.IO;

namespace Cybotech.Network
{
    public class ClientState : IDisposable
    {
        private int _id;
        private int _port;
        private IPAddress _ip;
        private IPEndPoint _endPoint;
        private Socket _socket;
        private ForwardStream _stream;
        private byte[] _buffer;

        public ClientState(IPEndPoint endPoint, Socket socket)
        {
            Init(endPoint, socket);
        }

        private void Init(IPEndPoint endPoint, Socket socket)
        {
            _endPoint = endPoint;
            _ip = _endPoint.Address;
            _port = _endPoint.Port;
            _id = endPoint.GetHashCode();
            _socket = socket;
            _stream = new ForwardStream();
            _buffer = new byte[Globals.BUFFER_SIZE];
        }

        public int Id
        {
            get { return _id; }
        }

        public int Port
        {
            get { return _port; }
        }

        public IPAddress Ip
        {
            get { return _ip; }
        }

        public IPEndPoint EndPoint
        {
            get { return _endPoint; }
        }

        public Socket Socket
        {
            get { return _socket; }
        }

        public ForwardStream Stream
        {
            get { return _stream; }
        }

        public byte[] Buffer
        {
            get { return _buffer; }
            set { _buffer = value; }
        }

        protected virtual void Dispose(bool disposing)
        {
            if (disposing)
            {
                if (_stream != null)
                {
                    _stream.Close();
                    _stream.Dispose();
                }

                if (_socket != null)
                {
                    _socket.Close();
                }
            }
        }

        public void Dispose()
        {
            Dispose(true);
        }
    }
}

using System;
using System.Collections.Generic;
using System.Net;
using System.Net.Sockets;
using Cybotech.Command;
using Cybotech.Network;

namespace ExamServer.Network
{
    public class TcpServer : IDisposable
    {

        private Socket socket;
        private bool secure;

        private readonly Dictionary<IPEndPoint, ClientState> clients = new Dictionary<IPEndPoint, ClientState>();

        //public events
        #region Events

        public event CommandDelegate CommandReceived;
        public event ConnectedDelegate ClientAdded;
        public event DisconnectedDelegate ClientRemoved;

        #endregion

        //event invokers
        #region Event Invoke methods

        protected virtual void OnCommandReceived(IPEndPoint ep, NetworkCommand command)
        {
            CommandDelegate handler = CommandReceived;
            if (handler != null) handler(ep, command);
        }

        protected virtual void OnClientAdded(IPEndPoint ep)
        {
            ConnectedDelegate handler = ClientAdded;
            if (handler != null) handler(ep);
        }

        protected virtual void OnClientDisconnect(IPEndPoint ep)
        {
            DisconnectedDelegate handler = ClientRemoved;
            if (handler != null) handler(ep);
        }

        #endregion

        //public property
        public string CertificatePath { get; set; }

        public TcpServer(EndPoint endPoint, bool secure)
        {
            StartServer(endPoint, secure);
        }

        public TcpServer(IPAddress ip, int port, bool secure)
        {
            StartServer(new IPEndPoint(ip, port), secure);
        }

        public TcpServer(string host, int port, bool secure)
        {
            StartServer(new IPEndPoint(IPAddress.Parse(host), port), secure);
        }

        private void StartServer(EndPoint ep, bool ssl)
        {
            socket = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp);
            socket.Bind(ep);
            socket.Listen(150);
            this.secure = ssl;

            socket.BeginAccept(AcceptClientCallback, null);
        }

        private void AcceptClientCallback(IAsyncResult ar)
        {
            Socket client = socket.EndAccept(ar);
            var ep = (IPEndPoint) client.RemoteEndPoint;
            var state = new ClientState(ep, client);
            if (secure)
            {
                //TODO : handle client for ssl authentication
            }

            //add client to 
            clients.Add(ep, state);
            OnClientAdded(ep);
            client.BeginReceive(state.Buffer, 0, state.Buffer.Length, SocketFlags.None, ReceiveDataCallback, state);

            //var thread = new Thread(ReceiveDataCallback);
            //thread.Start(state);
        }

        private void ReceiveDataCallback(IAsyncResult ar)
        {
            ClientState state = (ClientState)ar.AsyncState;

            try
            {
                var bytesRead = state.Socket.EndReceive(ar);
                state.Stream.Write(state.Buffer, 0, bytesRead);

                // check available commands
                while (state.Stream.LengthPrefix > 0)
                {
                    NetworkCommand cmd = NetworkCommand.CreateFromStream(state.Stream);
                    OnCommandReceived(state.EndPoint, cmd);
                }

                //start reading data again
                state.Socket.BeginReceive(state.Buffer, 0, state.Buffer.Length, SocketFlags.None, ReceiveDataCallback, state);
            }
            catch (SocketException ex)
            {
                if (ex.NativeErrorCode.Equals(10054))
                {
                    RemoveClient(state.EndPoint);
                }
            }
        }

        private void RemoveClient(IPEndPoint ep)
        {

            OnClientDisconnect(ep);
        }

        protected virtual void Dispose(bool disposing)
        {
            if (disposing)
            {
                //TODO : dispose all the client related socket stuff
            }
        }

        public void Dispose()
        {
            Dispose(true);
        }
    }
}

Answer 1

除非他完成当前字节的发送，否则同一客户端将无法向您发送数据。

所以在服务器端，您将收到完整的数据，不会被来自该客户端的其他新消息中断，但要考虑到如果发送的消息太大，并非所有发送的消息都会被一击接收，但仍然，接收完毕，最后是一条消息。

Answer 2

当您使用 TCP 时，网络协议将确保数据包的接收顺序与发送顺序相同。
关于线程安全，它在一定程度上取决于您用于发送的实际类。您提供的代码片段中缺少声明部分。
从名称上看，您似乎使用 Socket 并且这是线程安全的，因此每次发送实际上都是原子的，如果您使用 Stream 的任何风格，那么它不是线程安全的，您需要某种形式的同步，例如锁，您目前仍在使用。
如果您要发送大数据包，那么将接收和处理部分分成两个不同的线程很重要。 TCP 缓冲区实际上比人们想象的要小得多，不幸的是，当它已满时它不会被覆盖在日志中，因为协议将继续执行重新发送，直到收到所有内容。