c# - 在 C# 中的有限线程池上使用来自多个套接字的 IO 完成端口

Question

我正在尝试监听来自各种传入端口 (~20) 的 UDP 数据包。我想将 ~3-5 个线程用于接收和处理这些数据包。这似乎是 Windows 中 IO 完成端口的理想情况。我不明白的是如何对多个套接字进行从多到少的映射以检查较小的线程集。

以下代码创建我所有的套接字并开始异步接收操作。

for(int ix = 0; ix < 20; ix++)
{
    var socket = new Socket(AddressFamily.InterNetwork,
                            SocketType.Dgram, ProtocolType.Udp);
    socket.Bind(new IPEndPoint(IPAddress.Any, ix+6000));
    var e = new SocketAsyncEventArgs();
    e.Completed+=OnReceive;
    e.SetBuffer(buffer, ix*1024*1024, 1024*1024);
    socket.ReceiveFromAsync(e);
    _sockets.Add(socket);
}

我知道每个 OnReceive 消息都会在收到数据包时调用...

static void OnReceive(object sender, SocketAsyncEventArgs e)
{
    Console.WriteLine("Received {0} bytes", e.BytesTransfered);
    if(!((Socket)sender).ReceiveFromAsync(e))
        e_Completed(sender, e);
}

1. 如何限制运行 OnReceive 事件的线程数？
2. 在 OnReceive 方法递归调用自身次数过多的罕见情况下，防止堆栈溢出的最佳方法是什么？

Answer 1

不确定我是否理解接收和处理线程的意思。接收发生在后台。

无论如何，我会使用 BlockingCollection。

OnReceive 看起来像这样

private static BlockingCollection<byte[]> _received = new ...

static void OnReceive(object sender, SocketAsyncEventArgs e) {
    byte[] data = new byte[e.BytesTransfered];
    Array.Copy(e.buffer, e.Offset, data, 0, e.BytesTransfered);
    _received.Add(data)
    ...
}

然后您只需使用 TPL/PLinq 在请求的线程数上处理接收到的数据。

var parallelOptions = new ParalellOptions { MaxDegreeOfParallelism = 3 };
Parallel.ForEach(_received.GetConsumingPartitioner(),
                parallelOptions, 
                data => {
                    // do processing
                    ...
                });