3

我的目标是从套接字读取消息，其中每条消息都用 ETX 字符分隔。这是一个高频市场数据馈送，所以我认为逐字节的方法没有意义，而且完整消息的大小也是未知的。

有没有办法通过使用NetworkStream类来阅读此消息？我也尝试过Socket为此目的使用类，但不是从套接字中一一读取消息，而是从套接字读取所有消息，并且随着系统变慢，这成为一个问题。

4

3 回答 3

6

开始了; 这是从诸如 a或之类的源读取标记分隔的消息列表的基本过程。棘手的一点是跟踪您在传入缓冲区中使用的内容，以及来自早期缓冲区的未使用数据的任何积压。请注意，在和之间更改此代码本质上是更改为- 除了方法相同。SocketStreamSocketStreamReceiveRead

以下应该基本上可以满足您的需求。您可以使用ReadNext()API 直到获得 a null（表示流结束），也可以使用ReadAll()它为您提供IEnumerable<string>序列。编码和缓冲区大小可供您通过构造函数进行调整，但默认为正常值。

foreach (var s in reader.ReadAll())
    Console.WriteLine(s);

代码：

class EtxReader : IDisposable
{
    public IEnumerable<string> ReadAll()
    {
        string s;
        while ((s = ReadNext()) != null) yield return s;
    }
    public void Dispose()
    {
        if (socket != null) socket.Dispose();
        socket = null;
        if (backlog != null) backlog.Dispose();
        backlog = null;
        buffer = null;
        encoding = null;
    }
    public EtxReader(Socket socket, Encoding encoding = null, int bufferSize = 4096)
    {
        this.socket = socket;
        this.encoding = encoding ?? Encoding.UTF8;
        this.buffer = new byte[bufferSize];
    }
    private Encoding encoding;
    private Socket socket;
    int index, count;
    byte[] buffer;
    private bool ReadMore()
    {
        index = count = 0;
        int bytes = socket.Receive(buffer);
        if (bytes > 0)
        {
            count = bytes;
            return true;
        }
        return false;
    }
    public const byte ETX = 3;
    private MemoryStream backlog = new MemoryStream();
    public string ReadNext()
    {
        string s;
        if (count == 0)
        {
            if (!ReadMore()) return null;
        }
        // at this point, we expect there to be *some* data;
        // this may or may not include the ETX terminator
        var etxIndex = Array.IndexOf(buffer, ETX, index);
        if (etxIndex >= 0)
        {
            // found another message in the existing buffer
            int len = etxIndex - index;
            s = encoding.GetString(buffer, index, len);
            index = etxIndex + 1;
            count -= (len + 1);
            return s;
        }
        // no ETX in the buffer, so we'll need to fetch more data;
        // buffer the unconsumed data that we have
        backlog.SetLength(0);
        backlog.Write(buffer, index, count);

        bool haveEtx;
        do
        {
            if (!ReadMore())
            {
                // we had unused data; this must signal an error
                throw new EndOfStreamException();
            }
            etxIndex = Array.IndexOf(buffer, ETX, index);
            haveEtx = etxIndex >= 0;
            if (!haveEtx)
            {
                // keep buffering
                backlog.Write(buffer, index, count);
            }

        } while (!haveEtx);

        // now we have some data in the backlog, and the ETX in the buffer;
        // for convenience, copy the rest of the next message into
        // the backlog
        backlog.Write(buffer, 0, etxIndex);
        s = encoding.GetString(backlog.GetBuffer(), 0, (int)backlog.Length);
        index = etxIndex + 1;
        count -= (etxIndex + 1);
        return s;
    }
}
于 2013-06-13T12:38:48.990 回答
2

那么，这大概是一个基于文本的 API。在这里使用 aNetworkStream和 a没有实际区别Socket；既Stream不会也不会Socket“阅读所有消息” - 只有您的代码才能做到这一点。

在这两种情况下，您都需要一个几乎相同的循环来获取下一个数据块（这不是“消息”的同义词），并开始寻找您的哨兵值（您的意思是ETX？） - 根据需要进行处理或缓冲。除非您知道传入的提要采用单字节编码，否则最好将其视为字节，直到您实际将其拆分为逻辑消息，然后才对其运行文本解码器以获取这条消息，然后再转到下一条。

于 2013-06-13T11:49:24.903 回答
1

您应该研究异步通信和TcpListener类。我的方法是：

  1. 创建监听器
  2. 让它连续监听连接（BeginAccept/ EndAccecpt）。
  3. 对于每个连接，从 异步读取，NetworkStream直到客户端断开连接 ( BeginRead/ EndRead)。您可以读取数据块，例如您可以尝试一次读取 512 个字节 - 如果缓冲区中的字节数少于 512 个字节，那么您将获得少于 512 个字节的数据。
  4. 将任何内容附加到一个StringBuilder（每个连接一个，转换为时注意正确的byte[]编码string）
  5. 如果StringBuilder包含分隔符，则将该消息拆分并将其写入队列（不要忘记在入队之前锁定队列！）
  6. 有一个单独的线程持续监视该队列中的新消息并处理它们。如果您使用例如 a 将新内容放入队列中，您也可以向线程发出信号ManualResetEvent。

这只是一个粗略的大纲，但我相信你明白了。

没有读取“消息”之类的东西——通过 TCP/IP 传入的所有内容都只是字节流——这就是你得到网络流的原因。消息是您发明的用于解释传入数据的概念。

于 2013-06-13T11:50:19.423 回答