1

当我真正对我的网络代码进行压力测试时，我遇到了一些问题。基本上，一旦设置了套接字，它就会调用：

NetworkStream networkStream = mClient.GetStream();
networkStream.BeginRead(buffer, 0, buffer.Length, ReadCallback, buffer);


private void ReadCallback(IAsyncResult result)
        {
            try
            {
                int read;
                NetworkStream networkStream;
                try
                {
                    networkStream = mClient.GetStream();
                    read = networkStream.EndRead(result);
                }
                catch
                {
                    return;
                }

                if (read == 0)
                {
                    //The connection has been closed.
                    return;
                }

                var readBuffer = (byte[])result.AsyncState;
                var readCount = readBuffer.Length;
                while (readCount < 4)
                {
                    readCount += networkStream.Read(readBuffer, 0, readBuffer.Length - readCount);
                }
                var length = BitConverter.ToInt32(readBuffer, 0);
                var messageBuffer = new byte[length];
                readCount = 0;
                while (readCount < length)
                {
                    readCount += networkStream.Read(messageBuffer, 0, messageBuffer.Length - readCount);
                }
                else
                {
                    RaiseMessageReceived(this, messageBuffer);
                }
                //Then start reading from the network again.
                readBuffer = new byte[4]; //may not need to reset, not sure
                networkStream.BeginRead(readBuffer, 0, readBuffer.Length, ReadCallback, readBuffer);
            }
            catch(Exception)
            {
                //Connection is dead, stop trying to read and wait for a heal to retrigger the read queue
                return;
            }
        }

然后下面是我的发送方法

private byte[] GetMessageWithLength(byte[] bytes)
        {
            //Combine the msg length to the msg
            byte[] length = BitConverter.GetBytes(bytes.Length);
            var msg = new byte[length.Length + bytes.Length];
            Buffer.BlockCopy(length, 0, msg, 0, length.Length);
            Buffer.BlockCopy(bytes, 0, msg, length.Length, bytes.Length);
            return msg;
        }

public override bool Send(byte[] bytes)
        {
            lock (sendQueue)
            {
                sendQueue.Enqueue(bytes);
                Interlocked.Increment(ref sendQueueSize);
            }
            if (!mClient.Connected)
            {
                if (Connect())
                {
                    RaiseConnectionChanged(this, true, Localisation.TCPConnectionEstablished);
                }
                else
                {
                    RaiseConnectionChanged(this, false, (bytes.Length > 0 ? Localisation.TCPMessageFailed : Localisation.TCPMessageConnectionLost));
                }
            }

            try
            {
                NetworkStream networkStream = mClient.GetStream();

                lock (sendQueue)
                {
                    if (sendQueue.Count == 0)
                    {
                        return true;
                    }
                    bytes = sendQueue.Dequeue();
                }
                var msg = GetMessageWithLength(bytes);
                //Start async write operation
                networkStream.BeginWrite(msg, 0, msg.Length, WriteCallback, null);
            }
            catch (Exception ex)
            {
                RaiseConnectionChanged(this, false, (bytes.Length > 0 ? Localisation.TCPMessageFailed : Localisation.TCPMessageConnectionLost));
            }
            return true;
        }

        /// <summary>
        /// Callback for Write operation
        /// </summary>
        /// <param name="result">The AsyncResult object</param>
        private void WriteCallback(IAsyncResult result)
        {
            try
            {
                NetworkStream networkStream = mClient.GetStream();
                while (sendQueue.Count > 0)
                {
                    byte[] bytes;
                    lock (sendQueue)
                    {
                        if (sendQueue.Count == 0)
                        {
                            break;
                        }
                        bytes = sendQueue.Dequeue();
                    }
                    var msg = GetMessageWithLength(bytes);
                    networkStream.Write(msg, 0, msg.Length);
                    Interlocked.Decrement(ref sendQueueSize);
                }
                networkStream.EndWrite(result);
                mLastPacketSentAt = Environment.TickCount;
                Interlocked.Decrement(ref sendQueueSize);
            }
            catch (Exception ex)
            {
                RaiseConnectionChanged(this, false, Localisation.TCPMessageConnectionLost);
            }
        }

但是，是的，当我对系统进行压力测试时（比如 500 个左右的客户端同时发送大量消息），我注意到每 400 万个数据包中可能有 1 个数据包没有被接收。我不确定问题在于发送还是接收，这就是我包含这两种方法的原因。但是我会指出，如果我选择从客户端发送另一个数据包，它仍然可以正确发送和接收，所以它不仅仅是排队或其他东西。

谁能看到我缺少的东西？

4

1 回答 1

1

两个读取循环（例如while (readCount < length)）是错误的。您总是在零偏移处读取。您应该以不断增加的偏移量阅读。

这会导致已经读取的数据被覆盖。

另外，我不确定混合同步和异步读取是否是个好主意。这样你就失去了异步代码的好处，仍然必须处理回调等。我认为你应该决定一种风格并坚持下去。

于 2012-10-08T11:29:05.850 回答