C# - 从 TCP 连接读取时一些数据包被丢弃

Question

我正在使用 C# 通过 TCP 连接读取连续的数据流(ITCH 数据，即外汇价格)，但在长时间运行应用程序后，有时应用程序会丢弃一个数据包并且信息丢失。

下面是我用来读取数据的代码片段:

private void ReaderThreadStarter()
    {
        StreamReader streamReader = new StreamReader(this._networkStream);
        while (!_stopping)
        {
            try
            {
                if (this._networkStream.DataAvailable)
                {
                    while ((line = streamReader.ReadLine()) != null)
                    {
                        lock (_queue.ConcurrentQueue)
                        {
                            byte[] data = System.Text.Encoding.ASCII.GetBytes(line);
                            Log.Info("Data Added in Queue: " + Encoding.ASCII.GetString(data, 0, data.Length));
                            _queue.WriteToQueue(data);
                        }
                    }
                }
            }
            catch (Exception exception)
            {
                Log.Error(exception);
            }
            finally
            {
                SetStopped();
            }
        }
    }

上面这段代码的作用是从TCP连接中读取数据写入并发队列，然后另一个线程使用队列中的数据进行处理。所以基本上是一个简单的生产者-消费者问题。

生产者-消费者部分似乎工作正常，因为我在队列中写入的内容已被消费者使用。

一种选择是使用嗅探器并确认应用程序正在丢弃数据包，但我在无法使用嗅探器的环境中工作。我认为存在数据包丢失的原因是，对于我的一些外汇订单，我从来没有被取消过，而且我的价格下降了，数据提供商告诉我最后的定价是正确的。

我还记录了在保存到队列之前从 TCP 端口读取的数据，因此从日志中我假设数据在从连接读取时丢失了。

谁能告诉我这里可能做错了什么或者丢包的原因是什么。

下面是我的消费者代码的代码片段:

public void ReadQueue()
    {
        try
        {
            while (true)
            {
                {
                    byte[] data = _queue.ReadFromQueue();

                    Parse(data);
                }
            }
        }
        catch (Exception exception)
        {
            Log.Error(exception);
        }
    }

public byte[] ReadFromQueue()
    {
        try
        {
            byte[] data;
            lock (this) // Enter synchronization block
            {
                ConcurrentQueue.TryDequeue(out data);
            } 
            return data;
        }
        catch (Exception exception)
        {
            Log.Error(exception);
            return null;
        }
    }

Answer 1

有两件事引起了我的注意；首先是您对 DataAvailable 的使用。使用它几乎从来都不是正确的做法。这主要用于在同步和异步方法之间进行选择。例如，它不会告诉您是否有更多数据入站，并且可能会给出“误报”(因为您正在使用它并不意味着的东西)导致您的循环过早退出。 DataAvailable 仅告诉您数据当前在本地缓冲区中是否可用，这就是它告诉您的全部内容。

我感兴趣的第二件事是 data 是二进制还是文本。您正在使用 StreamReader 的事实表明是文本，但是……您为什么要将它重新编码回 byte[]？如果它是任意二进制文件，那么您不能将它作为文本处理——那将不起作用。当您通过 StreamReader 获取它时，您已经破坏了内容。如果它是基于文本的协议(protocol)，请不要对其重新编码:使用字符串队列(或类似的)。

一个不相关的说明...如果队列确实并发，您可能不需要同步访问。