从CF上的多个TCP连接连续读取 [英] Continously Reading from Multiple TCP Connections on CF

问题描述

我有一个简单的TCP服务器，该服务器能够侦听并接受端口上的多个连接.然后，它持续等待数据从其连接中读取.为了方便起见，它将TcpClient的包装器类称为ConnectedClient，并使用ConnectedClients的列表(字典)来跟踪所有连接.基本上是这样的:

I have a simple TCP server that is able to listen for and accept multiple connections on a port. It then continuously waits for data to read from its connections. It uses a wrapper class for a TcpClient called ConnectedClient for convenience, and a list (dictionary) of ConnectedClients to keep track of all the connections. It basically goes like this:

/* this method waits to accept connections indefinitely until it receives 
   the signal from the GUI thread to stop. When a connection is accepted, it 
   adds the connection to the list and calls a method called ProcessClient, 
   which returns almost immediately.*/
public void waitForConnections() {
        // this method has access to a TcpListener called listener that was started elsewhere
        try {
            while (!_abort) {
                TcpClient socketClient = listener.AcceptTcpClient();

                //Connected client constructor takes the TcpClient as well as a callback that it uses to print status messages to the GUI if   
                ConnectedClient client = new ConnectedClient(socketClient, onClientUpdate);
                clients.Add(client.id, client);
                ProcessClient(client);
            }
        }
        catch (Exception e) {
            onStatusUpdate("Exception Occurred: " + e.Message);
        }
    }

    /* This method doesn't do much other than call BeginRead on the connection */
    private void ProcessClient(ConnectedClient client) {
        try {
            // wrapper class contains an internal buffer for extracting data as well as a TcpClient
            NetworkStream stream = client.tcpClient.GetStream();
            stream.BeginRead(client.buffer, 0, client.tcpClient.ReceiveBufferSize, new AsyncCallback(StreamReadCompleteCallback), client);
        }
        catch (Exception ex) {
            onStatusUpdate(ex.Message);
        }
    }

在我的回调函数StreamReadCompleteCallback中，我调用EndRead，检查EndRead的返回值以检测连接是否已关闭.如果返回值大于零，则提取/处理读取的数据，然后在同一客户端上再次调用BeginRead.如果返回值为零，则说明连接已关闭，我删除了该连接(从列表中删除，关闭TcpClient，等等).

In my callback function, StreamReadCompleteCallback, I call EndRead, checking the return value of EndRead to detect whether the connection has been closed. If the return value is greater than zero, I extract/process the read data and call BeginRead again on the same client. If the return value is zero, the connection has been closed and I remove the connection (delete from list, close the TcpClient, etc).

    private void StreamReadCompleteCallback(IAsyncResult ar) {
        ConnectedClient client = (ConnectedClient)ar.AsyncState;

        try {
            NetworkStream stream = client.tcpClient.GetStream();

            int read = stream.EndRead(ar);
            if (read != 0) {
                // data extraction/light processing of received data
                client.Append(read);
                stream.BeginRead(client.buffer, 0, client.tcpClient.ReceiveBufferSize, new AsyncCallback(StreamReadCompleteCallback), client);
            }
            else {
                DisconnectClient(client);
            }
        }
        catch (Exception ex) {
            onStatusUpdate(ex.Message);
        }
    }

所有这些都可以正常工作，我可以接受连接并从多个客户端设备等中读取信息.

All of this works fine, I can accept connections and read from multiple client devices, etc.

我的问题是:这种从连接的客户端连续读取数据的方法会导致每个连接都有一个等待BeginRead返回的辅助线程.

因此，如果我有10个连接，则有10个BeginReads.

So if I have 10 connections, I have 10 BeginReads going.

让如此多的工作线程围着等待阅读似乎很浪费.还有其他更好的方法可以做到这一点吗?如果活动连接数量很多，最终会耗尽内存以添加连接.

是否有一个线程会轮询每个连接的DataAvailable属性，直到出现某种情况，然后使要读取/处理的线程成为解决方案?

Would having a thread that polls the DataAvailable property of each connection until something shows up, and then makes a thread to read/process be a solution?

或者创建所有这些工作线程不是我想的那么重要吗?

推荐答案

这种从连接的客户端连续读取的方法导致每个连接都有一个等待BeginRead返回的工作线程.

This method of continuously reading from connected clients causes each connection to have a worker thread that is waiting for BeginRead to return

不，不是.实际上，使用BeginRead()或其他异步替代方法之一来处理Socket对象上的I/O是最可扩展的方法.

No, it doesn't. In fact, using BeginRead() or one of the other asynchronous alternatives to processing I/O on a Socket object is the most scalable approach to use.

是否有一个线程会轮询每个连接的DataAvailable属性，直到出现某种情况，然后使要读取/处理的线程成为解决方案?

Would having a thread that polls the DataAvailable property of each connection until something shows up, and then makes a thread to read/process be a solution?

不.这太可怕了.通过DataAvailable或Select()轮询套接字效率极低，仅检查套接字状态就不得不花费大量CPU时间.操作系统提供了良好的异步机制来处理此问题.轮询实现会忽略该问题，并自行完成所有工作.

No. This would be horrible. Polling a socket, via DataAvailable or Select(), is terribly inefficient, forcing a huge amount of CPU time to be invested just checking on the socket state. The OS provides good asynchronous mechanisms for handling this; a polling implementation ignores that and does all the work itself.

或者创建所有这些工作线程不是我想的那么重要吗?

Or is creating all these worker threads not as big of a deal as I think?

您不是在创建自己认为的线程.当您使用异步API时，它们会利用Windows中称为 I/O完成端口的功能. I/O完成端口与I/O操作相关联，线程可以在端口上等待.但是一个线程可以处理大量操作的等待，因此拥有十个未完成的读取操作实际上并不会导致创建十个不同的线程.

You aren't creating the threads you think you are. When you use the asynchronous APIs, they make use of a feature in windows called I/O Completion Ports. An I/O Completion Port is associated with an I/O operation, and a thread can wait on a port. But one thread can handle waiting on a large number of operations, so having ten outstanding read operations does not in fact cause ten different threads to be created.

.NET管理线程池以处理这些操作，这些线程作为ThreadPool类的一部分进行管理.您可以监视该类以查看IOCP池的行为(与用于QueueUserWorkItem()的工作线程池不同).

.NET manages a pool of threads to handle these operations, managed as part of the ThreadPool class. You can monitor that class to see the behavior of the IOCP pool (which is different from the worker thread pool used for QueueUserWorkItem()).

.NET将根据需要分配新的IOCP对象和线程，以服务于您的网络I/O操作.您可以放心，它将以合理，有效的方式做到这一点.

.NET will assign new IOCP objects and threads as needed to service your network I/O operations. You can rest assured that it will do so in a reasonable, efficient manner.

在非常大的规模上，与读取操作关联的对象的垃圾回收开销可能会发挥作用.在这种情况下，可以使用ReceiveAsync()方法，该方法允许您将自己的状态对象池重新用于操作，这样就不必不断创建和丢弃对象.

At very large scales, the overhead of the garbage collection of the objects associated with read operations may come into play. In this case, you can use the ReceiveAsync() method, which allows you to reuse your own pool of state objects for the operations, so that you aren't constantly creating and discarding objects.

另一个可能出现的问题是内存碎片，特别是在大对象堆中(取决于您使用的缓冲区的大小).在套接字上开始读取操作时，必须固定缓冲区，以防止.NET压缩它所驻留的堆.

Another issue that may come up is memory fragmentation, especially in the large-object heap (depending on the size of the buffers you use). When you start a read operation on a socket, the buffer has to be pinned, preventing .NET from compacting the heap in which it resides.

但是这些问题并不是避免使用异步API的原因(实际上，第二个问题无论如何都会发生).它们只是要注意的事情.实际上，使用异步API是最好的方法.

But these issues aren't reasons to avoid using the asynchronous APIs (and in fact, the second issue happens regardless). They are just things to be aware of. Using the asynchronous API is in fact the best way to go.

也就是说，BeginReceive()是老派".它可以工作，但是您可以将BeginReceive()操作包装在Task中(请参阅

That said, BeginReceive() is "old school". It works, but you can wrap a BeginReceive() operation in a Task (see Task.FromAsync() and TPL and Traditional .NET Framework Asynchronous Programming), or you can wrap the entire Socket in a NetworkStream object (which has ReadAsync() and similar methods), which will allow you to write your asynchronous code in a more readable way that doesn't require the use of explicit callback methods. And for scenarios where the network I/O always culminates in some interaction with the UI, allows you to use async/await to do so, again in a more readable, easier-to-write way.

这篇关于从CF上的多个TCP连接连续读取的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！