C# 中的高性能 TCP 服务器答案

【问题标题】：High performance TCP server in C#C# 中的高性能 TCP 服务器
【发布时间】：2011-08-26 18:24:14
【问题描述】：

我是一位经验丰富的 C# 开发人员，但到目前为止我还没有开发过 TCP 服务器应用程序。现在我必须开发一个高度可扩展和高性能的服务器，它可以处理至少 5-10,000 个并发连接：通过 GPRS 从 GPS 设备获取 -raw byte-data。

一个常见的沟通过程应该是这样的：

GPS 设备启动与我的服务器的连接
如果我想获取数据，我的服务器会回答
设备发送 GPS 数据
我的服务器向设备发送有关获取它的报告（例如校验和）
从 GPS、reportm 获取新数据，这种情况一次又一次地发生
稍后 GPS DEVICE 关闭连接

所以，在我的服务器中我需要

跟踪已连接/活动的客户端
从服务器端关闭任何客户端
在设备关闭连接时捕获事件
获取字节数据
向客户发送数据

我开始在互联网上阅读有关此主题的信息，但这对我来说似乎是一场噩梦。有很多方法，但我找不到最好的方法。

异步套接字方法对我来说似乎是最好的，但是以这种异步风格编写代码很糟糕而且不容易调试。

所以我的问题是：您认为用 C# 实现高性能 TCP 服务器的最佳方式是什么？你知道有什么好的开源组件可以做到这一点吗？（我试了好几个，都没有找到好的。）

【问题讨论】：

不确定我是否理解流程。 “我的服务器向设备发送有关获取它的报告”是什么意思？另外，为什么需要保持连接打开？为什么不在必要时打开它们？
您能否详细说明为什么您说“以这种异步方式编写代码很糟糕且不易调试”？
因此，它必须保持打开状态，因为它们是 GPRS 连接和 GPS，当它们连接时 - 主要是在漫游时 - 需要花钱。所以一切都是为了钱，更多的连接=更多的费用。报告意味着：服务器必须向它获取数据的设备发送报告。（它们是一些字节）当设备得到这个报告时，它可以从它的内存中删除发送的数据，如果它没有这个报告，则尝试再次发送这个口袋。
我正在做同一个项目。（我认为您正在构建 GPS 跟踪系统或类似的东西，不是吗？）。我会和你分享经验。 StackExchange 上没有私人消息传递。可以联系我吗？地址：（我的昵称 AT gmail.com）（在这篇文章被删除之前很快：p）谢谢。

标签： c# tcp scalable

【解决方案1】：

它必须是异步的，没有办法解决这个问题。高性能和可扩展性不能与每个套接字一个线程混在一起。您可以查看 StackExchange 自己在做什么，请参阅 async Redis await BookSleeve，它利用了下一个 C# 版本中的 CTP 功能（因此处于边缘并且可能会发生变化，但它很酷）。对于更前沿的解决方案，解决方案围绕利用 SocketAsyncEventArgs Class 发展，通过消除与“经典”C# 异步处理相关的异步处理程序的频繁分配，使事情更进一步：

SocketAsyncEventArgs 类是一部分的一组增强功能 System.Net.Sockets.Socket 类提供替代异步可以使用的模式专用高性能插座应用程序。这堂课是专为网络设计要求高的服务器应用程序表现。一个应用程序可以使用增强的异步模式专门或仅在有针对性的热点区域（例如，当接收大量数据）。

长话短说：学习异步或尝试死亡......

顺便说一句，如果你问为什么异步，那么请阅读这篇文章链接的三篇文章：High Performance Windows programs。最终答案是：底层操作系统设计需要它。

【讨论】：

为每个客户端创建一个线程不是同样有效吗？我的意思是，当您调用异步方法时，您实际上是在启动一个线程，该线程在完成时调用委托，或者至少我是这样理解的。此外，如果这是真的，您最终可能会获得更高的线程数，这可能不是最好的性能。如果这是正确的，那么它必须在每次调用时创建一个新线程，其中单个线程循环通过标准阻塞方法似乎更有效。
@kelton52：您的理解完全错误。你需要了解异步 IO。
@RemusRusanu 如果我的理解完全错误，那么为什么在我运行异步任务时，它们会创建线程？如何在不打开线程的情况下异步运行任务？我认为在这种情况下，除了“您的理解完全错误”之外的评论是值得的，因为这不是一个反驳的论点。
您必须了解如何在最低级别（内核/驱动程序）处理异步 io，以了解为什么它不需要为您发布的每个 IO 使用线程。在这个狭小的空间里，我能让你走上正轨的最简单方法是指出内核深处只有 1 个“线程”控制一切（泵送所有线程），如果你从那时开始考虑它看来，您可以看到异步 IO 是如何由于操作系统本身而优化的。也许阅读“线程上下文切换”。
“端口耗尽”与服务器在端口上侦听和接受（无论有多少）客户端连接无关。这只是启动许多连接（出站）的进程的问题，即在短时间内启动与服务器的许多连接的客户端。这就是为什么您的链接谈论the default range of ephemeral ports used by *client applications*。另请查看this technet article 并搜索“出站”。

【解决方案2】：

你可以使用TcpClient 类来做到这一点，虽然说实话我不知道你是否可以有 10000 个打开的套接字。这是相当多的。但我经常使用TcpClient 来处理几十个并发套接字。而且异步模型其实很好用。

你最大的问题不是让TcpClient 工作。有 10,000 个并发连接，我认为带宽和可扩展性将成为问题。我什至不知道一台机器是否可以处理所有这些流量。我想这取决于数据包有多大以及它们进入的频率。但在你承诺在一台计算机上实现这一切之前，你最好做一些粗略的估计。

【讨论】：

这个口袋很小，每个口袋 1024 字节。多常？这取决于 GPS 设备的设置，取决于移动速度、航向等。
问题标题提到了 TCP 服务器。您的意思是说 TCPListener 或者您的答案的性质不正确。即使数据最终会从一个或多个源 GPS 分发到一个或多个消费客户端，发布者和订阅者都可能会连接到 TCP 服务器以发布或接收数据。

【解决方案3】：

正如 Remus 上面所说，您必须使用异步来保持高性能。这就是 .NET 中的 Begin.../End... 方法。

在套接字的底层，这些方法利用了 IO 完成端口，这似乎是在 Windows 操作系统上处理许多套接字的最高效方式。

正如 Jim 所说，TcpClient 类在这里可以提供帮助并且非常易于使用。这是一个使用 TcpListener 监听传入连接并使用 TcpClient 处理它们的示例，初始 BeginAccept 和 BeginRead 调用是异步的。

此示例确实假设在套接字上使用了基于消息的协议，并且除了每次传输的前 4 个字节是长度外，该协议被省略，但随后允许您在流上使用同步读取来获取其余部分已经缓冲的数据。

代码如下：

class ClientContext
{
    public TcpClient Client;
    public Stream Stream;
    public byte[] Buffer = new byte[4];
    public MemoryStream Message = new MemoryStream();
}

class Program
{
    static void OnMessageReceived(ClientContext context)
    {
        // process the message here
    }

    static void OnClientRead(IAsyncResult ar)
    {
        ClientContext context = ar.AsyncState as ClientContext;
        if (context == null)
            return;

        try
        {
            int read = context.Stream.EndRead(ar);
            context.Message.Write(context.Buffer, 0, read);

            int length = BitConverter.ToInt32(context.Buffer, 0);
            byte[] buffer = new byte[1024];
            while (length > 0)
            {
                read = context.Stream.Read(buffer, 0, Math.Min(buffer.Length, length));
                context.Message.Write(buffer, 0, read);
                length -= read;
            }

            OnMessageReceived(context);
        }
        catch (System.Exception)
        {
            context.Client.Close();
            context.Stream.Dispose();
            context.Message.Dispose();
            context = null;
        }
        finally
        {
            if (context != null)
                context.Stream.BeginRead(context.Buffer, 0, context.Buffer.Length, OnClientRead, context);
        }
    }

    static void OnClientAccepted(IAsyncResult ar)
    {
        TcpListener listener = ar.AsyncState as TcpListener;
        if (listener == null)
            return;

        try
        {
            ClientContext context = new ClientContext();
            context.Client = listener.EndAcceptTcpClient(ar);
            context.Stream = context.Client.GetStream();
            context.Stream.BeginRead(context.Buffer, 0, context.Buffer.Length, OnClientRead, context);
        }
        finally
        {
            listener.BeginAcceptTcpClient(OnClientAccepted, listener);
        }
    }

    static void Main(string[] args)
    {
        TcpListener listener = new TcpListener(new IPEndPoint(IPAddress.Any, 20000));
        listener.Start();

        listener.BeginAcceptTcpClient(OnClientAccepted, listener);

        Console.Write("Press enter to exit...");
        Console.ReadLine();
        listener.Stop();
    }
}

它演示了如何处理异步调用，但需要添加错误处理以确保 TcpListener 始终接受新连接，并在客户端意外断开连接时进行更多错误处理。此外，在某些情况下，并非所有数据都一次性到达，也需要处理。

【讨论】：

【解决方案4】：

我认为您也在寻找 UDP 技术。对于 10k 客户端，它很快，但问题是您必须为收到消息的每条消息实施确认。在 UDP 中，您不需要为每个客户端打开一个套接字，但需要在 x 秒后实现心跳/ping 机制来检查哪个客户端已连接。

【讨论】：

同意。这个项目（现在已经完成了，对吧？）应该被视为一个“软”实时应用程序。现在我唯一想弄清楚的是为什么这一切都必须在一台机器上运行。

【解决方案5】：

您可以使用我制作的 TCP CSharpServer，实现起来非常简单，只需在您的一个类上实现 IClientRequest 即可。

using System;
using System.Collections.Generic;
using System.Linq;

namespace cSharpServer
{
    public interface IClientRequest
    {        
        /// <summary>
        /// this needs to be set, otherwise the server will not beable to handle the request.
        /// </summary>
        byte IdType { get; set; } // This is used for Execution.
        /// <summary>
        /// handle the process by the client.
        /// </summary>
        /// <param name="data"></param>
        /// <param name="client"></param>
        /// <returns></returns>
        byte[] Process(BinaryBuffer data, Client client);
    }
}

BinaryBuffer 让你读取发送到服务器的数据真的很容易。

using System;
using System.Collections.Generic;
using System.Linq;
using System.Runtime.CompilerServices;
using System.Text;
using System.Threading.Tasks;

namespace cSharpServer
{
    public class BinaryBuffer
    {
        private const string Str0001 = "You are at the End of File!";
        private const string Str0002 = "You are Not Reading from the Buffer!";
        private const string Str0003 = "You are Currenlty Writing to the Buffer!";
        private const string Str0004 = "You are Currenlty Reading from the Buffer!";
        private const string Str0005 = "You are Not Writing to the Buffer!";
        private const string Str0006 = "You are trying to Reverse Seek, Unable to add a Negative value!";
        private bool _inRead;
        private bool _inWrite;
        private List<byte> _newBytes;
        private int _pointer;
        public byte[] ByteBuffer;

        [MethodImpl(MethodImplOptions.AggressiveInlining)]
        public override string ToString()
        {
            return Helper.DefaultEncoding.GetString(ByteBuffer, 0, ByteBuffer.Length);
        }

        [MethodImpl(MethodImplOptions.AggressiveInlining)]
        public BinaryBuffer(string data)
            : this(Helper.DefaultEncoding.GetBytes(data))
        {
        }

        [MethodImpl(MethodImplOptions.AggressiveInlining)]
        public BinaryBuffer()
        {
        }

        [MethodImpl(MethodImplOptions.AggressiveInlining)]
        public BinaryBuffer(byte[] data)
            : this(ref data)
        {
        }

        [MethodImpl(MethodImplOptions.AggressiveInlining)]
        public BinaryBuffer(ref byte[] data)
        {
            ByteBuffer = data;
        }

        [MethodImpl(MethodImplOptions.AggressiveInlining)]
        public void IncrementPointer(int add)
        {
            if (add < 0)
            {
                throw new Exception(Str0006);
            }
            _pointer += add;
            if (EofBuffer())
            {
                throw new Exception(Str0001);
            }
        }

        [MethodImpl(MethodImplOptions.AggressiveInlining)]
        public int GetPointer()
        {
            return _pointer;
        }

        [MethodImpl(MethodImplOptions.AggressiveInlining)]
        public static string GetString(ref byte[] buffer)
        {
            return Helper.DefaultEncoding.GetString(buffer, 0, buffer.Length);
        }

        [MethodImpl(MethodImplOptions.AggressiveInlining)]
        public static string GetString(byte[] buffer)
        {
            return GetString(ref buffer);
        }

        [MethodImpl(MethodImplOptions.AggressiveInlining)]
        public void BeginWrite()
        {
            if (_inRead)
            {
                throw new Exception(Str0004);
            }
            _inWrite = true;

            _newBytes = new List<byte>();
        }

        [MethodImpl(MethodImplOptions.AggressiveInlining)]
        public void Write(float value)
        {
            if (!_inWrite)
            {
                throw new Exception(Str0005);
            }
            _newBytes.AddRange(BitConverter.GetBytes(value));
        }

        [MethodImpl(MethodImplOptions.AggressiveInlining)]
        public void Write(byte value)
        {
            if (!_inWrite)
            {
                throw new Exception(Str0005);
            }
            _newBytes.Add(value);
        }

        [MethodImpl(MethodImplOptions.AggressiveInlining)]
        public void Write(int value)
        {
            if (!_inWrite)
            {
                throw new Exception(Str0005);
            }

            _newBytes.AddRange(BitConverter.GetBytes(value));
        }

        [MethodImpl(MethodImplOptions.AggressiveInlining)]
        public void Write(long value)
        {
            if (!_inWrite)
            {
                throw new Exception(Str0005);
            }
            byte[] byteArray = new byte[8];

            unsafe
            {
                fixed (byte* bytePointer = byteArray)
                {
                    *((long*)bytePointer) = value;
                }
            }

            _newBytes.AddRange(byteArray);
        }

        [MethodImpl(MethodImplOptions.AggressiveInlining)]
        public int UncommitedLength()
        {
            return _newBytes == null ? 0 : _newBytes.Count;
        }

        [MethodImpl(MethodImplOptions.AggressiveInlining)]
        public void WriteField(string value)
        {
            Write(value.Length);
            Write(value);
        }

        [MethodImpl(MethodImplOptions.AggressiveInlining)]
        public void Write(string value)
        {
            if (!_inWrite)
            {
                throw new Exception(Str0005);
            }
            byte[] byteArray = Helper.DefaultEncoding.GetBytes(value);
            _newBytes.AddRange(byteArray);
        }

        [MethodImpl(MethodImplOptions.AggressiveInlining)]
        public void Write(decimal value)
        {
            if (!_inWrite)
            {
                throw new Exception(Str0005);
            }
            int[] intArray = decimal.GetBits(value);

            Write(intArray[0]);
            Write(intArray[1]);
            Write(intArray[2]);
            Write(intArray[3]);
        }

        [MethodImpl(MethodImplOptions.AggressiveInlining)]
        public void SetInt(int value, int pos)
        {
            byte[] byteInt = BitConverter.GetBytes(value);
            for (int i = 0; i < byteInt.Length; i++)
            {
                _newBytes[pos + i] = byteInt[i];
            }
        }

        [MethodImpl(MethodImplOptions.AggressiveInlining)]
        public void SetLong(long value, int pos)
        {
            byte[] byteInt = BitConverter.GetBytes(value);
            for (int i = 0; i < byteInt.Length; i++)
            {
                _newBytes[pos + i] = byteInt[i];
            }
        }

        [MethodImpl(MethodImplOptions.AggressiveInlining)]
        public void Write(byte[] value)
        {
            Write(ref value);
        }

        [MethodImpl(MethodImplOptions.AggressiveInlining)]
        public void Write(ref byte[] value)
        {
            if (!_inWrite)
            {
                throw new Exception(Str0005);
            }
            _newBytes.AddRange(value);
        }

        [MethodImpl(MethodImplOptions.AggressiveInlining)]
        public void EndWrite()
        {
            if (ByteBuffer != null)
            {
                _newBytes.InsertRange(0, ByteBuffer);
            }
            ByteBuffer = _newBytes.ToArray();
            _newBytes = null;
            _inWrite = false;
        }

        [MethodImpl(MethodImplOptions.AggressiveInlining)]
        public void EndRead()
        {
            _inRead = false;
            _pointer = 0;
        }

        [MethodImpl(MethodImplOptions.AggressiveInlining)]
        public void BeginRead()
        {
            if (_inWrite)
            {
                throw new Exception(Str0003);
            }
            _inRead = true;
            _pointer = 0;
        }

        [MethodImpl(MethodImplOptions.AggressiveInlining)]
        public byte ReadByte()
        {
            if (!_inRead)
            {
                throw new Exception(Str0002);
            }
            if (EofBuffer())
            {
                throw new Exception(Str0001);
            }
            return ByteBuffer[_pointer++];
        }

        [MethodImpl(MethodImplOptions.AggressiveInlining)]
        public int ReadInt()
        {
            if (!_inRead)
            {
                throw new Exception(Str0002);
            }
            if (EofBuffer(4))
            {
                throw new Exception(Str0001);
            }
            int startPointer = _pointer;
            _pointer += 4;

            return BitConverter.ToInt32(ByteBuffer, startPointer);
        }

        [MethodImpl(MethodImplOptions.AggressiveInlining)]
        public float[] ReadFloatArray()
        {
            float[] dataFloats = new float[ReadInt()];
            for (int i = 0; i < dataFloats.Length; i++)
            {
                dataFloats[i] = ReadFloat();
            }
            return dataFloats;
        }

        [MethodImpl(MethodImplOptions.AggressiveInlining)]
        public float ReadFloat()
        {
            if (!_inRead)
            {
                throw new Exception(Str0002);
            }
            if (EofBuffer(sizeof(float)))
            {
                throw new Exception(Str0001);
            }
            int startPointer = _pointer;
            _pointer += sizeof(float);

            return BitConverter.ToSingle(ByteBuffer, startPointer);
        }

        [MethodImpl(MethodImplOptions.AggressiveInlining)]
        public decimal ReadDecimal()
        {
            if (!_inRead)
            {
                throw new Exception(Str0002);
            }
            if (EofBuffer(16))
            {
                throw new Exception(Str0001);
            }
            return new decimal(new[] { ReadInt(),
                ReadInt(),
                ReadInt(),
                ReadInt()
            });
        }

        [MethodImpl(MethodImplOptions.AggressiveInlining)]
        public long ReadLong()
        {
            if (!_inRead)
            {
                throw new Exception(Str0002);
            }
            if (EofBuffer(8))
            {
                throw new Exception(Str0001);
            }
            int startPointer = _pointer;
            _pointer += 8;

            return BitConverter.ToInt64(ByteBuffer, startPointer);
        }

        [MethodImpl(MethodImplOptions.AggressiveInlining)]
        public string ReadString(int size)
        {
            return Helper.DefaultEncoding.GetString(ReadByteArray(size), 0, size);
        }

        [MethodImpl(MethodImplOptions.AggressiveInlining)]
        public byte[] ReadByteArray(int size)
        {
            if (!_inRead)
            {
                throw new Exception(Str0002);
            }
            if (EofBuffer(size))
            {
                throw new Exception(Str0001);
            }
            byte[] newBuffer = new byte[size];

            Array.Copy(ByteBuffer, _pointer, newBuffer, 0, size);

            _pointer += size;

            return newBuffer;
        }

        [MethodImpl(MethodImplOptions.AggressiveInlining)]
        public bool EofBuffer(int over = 1)
        {
            return ByteBuffer == null || ((_pointer + over) > ByteBuffer.Length);
        }
    }
}

完整项目在 GitHub CSharpServer

【讨论】：