使用C#进行TCP/IP网络编程

Question

大家好，

我要写一些代码来监听通过GPRS从GSM手机发来的TCPIP消息。最终，我希望这个程序能在一个虚拟私有服务器上运行，并且可能每秒处理多个消息。

我对网络编程不太熟悉，所以我在互联网上做了一些研究并阅读了一些教程。目前我考虑的方法是使用Windows服务和套接字来监听端口。如果我的理解正确，我需要一个套接字来监听客户端的连接，每次有人尝试连接该端口时，我将获得另一个套接字与他们进行通信。对于更有经验的人来说，这听起来是否正确？

我计划使用异步通信，但更大的设计问题之一是是否使用线程。线程不是我真正玩过的东西，我知道有很多陷阱——比如竞态条件和调试问题。

如果我避免使用线程，我知道我必须提供一个作为特定会话标识符的对象。我正在考虑使用GUIDs - 你有什么意见吗？

非常感谢任何回复...

Martin

- Martin Milan

---

你可能需要使用线程（或线程池）来为客户端请求提供服务。至少，在我的服务发现工作中就是这样的经验。 - jasonco

不，我们不需要。从.NET Framework 2.0 SP1开始，套接字实现基于IO完成端口。这种方法比使用一个线程处理一个传入连接要高效得多。我们不需要事件线程池。（有关更多信息，请参见我的答案）。 - Sergey Teplyakov

Answer 1

从.net framework 2.0 SP1开始，与异步套接字相关的套接字库发生了一些变化。

在内部使用了多线程。我们不需要手动使用多线程（甚至无需显式地使用ThreadPool）。我们所做的一切就是使用BeginAcceptSocket来开始接受新连接，并在接受新连接后使用SocketAsyncEventArgs。

简短实现：

//In constructor or in method Start
var tcpServer = new TcpListener(IPAddress.Any, port);
tcpServer.Start();
tcpServer.BeginAcceptSocket(EndAcceptSocket, tcpServer);

//In EndAcceptSocket
Socket sock= lister.EndAcceptSocket(asyncResult);
var e = new SocketAsyncEventArgs();
e.Completed += ReceiveCompleted; //some data receive handle
e.SetBuffer(new byte[SocketBufferSize], 0, SocketBufferSize);
if (!sock.ReceiveAsync(e))
{//IO operation finished syncronously
  //handle received data
  ReceiveCompleted(sock, e);
}//IO operation finished syncronously
//Add sock to internal storage

完整实现：

using System;
using System.Collections.Generic;
using System.Net;
using System.Net.Sockets;
using System.Runtime.InteropServices;

namespace Ample
{
    public class IPEndPointEventArgs : EventArgs
    {
        public IPEndPointEventArgs(IPEndPoint ipEndPoint)
        {
            IPEndPoint = ipEndPoint;
        }

        public IPEndPoint IPEndPoint { get; private set; }
    }

    public class DataReceivedEventArgs : EventArgs
    {
        public DataReceivedEventArgs(byte[] data, IPEndPoint ipEndPoint)
        {
            Data = data;
            IPEndPoint = ipEndPoint;
        }

        public byte[] Data { get; private set; }
        public IPEndPoint IPEndPoint { get; private set; }

    }
    /// <summary>
    /// TcpListner wrapper
    /// Encapsulates asyncronous communications using TCP/IP.
    /// </summary>
    public sealed class TcpServer : IDisposable
    {
        //----------------------------------------------------------------------
        //Construction, Destruction
        //----------------------------------------------------------------------
        /// <summary>
        /// Creating server socket
        /// </summary>
        /// <param name="port">Server port number</param>
        public TcpServer(int port)
        {
            connectedSockets = new Dictionary<IPEndPoint, Socket>();
            tcpServer = new TcpListener(IPAddress.Any, port);
            tcpServer.Start();
            tcpServer.BeginAcceptSocket(EndAcceptSocket, tcpServer);
        }
        ~TcpServer()
        {
            DisposeImpl(false);
        }
        public void Dispose()
        {
            DisposeImpl(true);
        }

        //----------------------------------------------------------------------
        //Public Methods
        //----------------------------------------------------------------------

        public void SendData(byte[] data, IPEndPoint endPoint)
        {
            Socket sock;
            lock (syncHandle)
            {
                if (!connectedSockets.ContainsKey(endPoint))
                    return;
                sock = connectedSockets[endPoint];
            }
            sock.Send(data);
        }

        //----------------------------------------------------------------------
        //Events
        //----------------------------------------------------------------------
        public event EventHandler<IPEndPointEventArgs> SocketConnected;
        public event EventHandler<IPEndPointEventArgs> SocketDisconnected;
        public event EventHandler<DataReceivedEventArgs> DataReceived;

        //----------------------------------------------------------------------
        //Private Functions
        //----------------------------------------------------------------------
        #region Private Functions
        //Обработка нового соединения
        private void Connected(Socket socket)
        {
            var endPoint = (IPEndPoint)socket.RemoteEndPoint;

            lock (connectedSocketsSyncHandle)
            {
                if (connectedSockets.ContainsKey(endPoint))
                {
                    theLog.Log.DebugFormat("TcpServer.Connected: Socket already connected! Removing from local storage! EndPoint: {0}", endPoint);
                    connectedSockets[endPoint].Close();
                }

                SetDesiredKeepAlive(socket);
                connectedSockets[endPoint] = socket;
            }

            OnSocketConnected(endPoint);
        }

        private static void SetDesiredKeepAlive(Socket socket)
        {
            socket.SetSocketOption(SocketOptionLevel.Socket, SocketOptionName.KeepAlive, true);
            const uint time = 10000;
            const uint interval = 20000;
            SetKeepAlive(socket, true, time, interval);
        }
        static void SetKeepAlive(Socket s, bool on, uint time, uint interval)
        {
            /* the native structure
            struct tcp_keepalive {
            ULONG onoff;
            ULONG keepalivetime;
            ULONG keepaliveinterval;
            };
            */

            // marshal the equivalent of the native structure into a byte array
            uint dummy = 0;
            var inOptionValues = new byte[Marshal.SizeOf(dummy) * 3];
            BitConverter.GetBytes((uint)(on ? 1 : 0)).CopyTo(inOptionValues, 0);
            BitConverter.GetBytes((uint)time).CopyTo(inOptionValues, Marshal.SizeOf(dummy));
            BitConverter.GetBytes((uint)interval).CopyTo(inOptionValues, Marshal.SizeOf(dummy) * 2);
            // of course there are other ways to marshal up this byte array, this is just one way

            // call WSAIoctl via IOControl
            int ignore = s.IOControl(IOControlCode.KeepAliveValues, inOptionValues, null);

        }
        //socket disconnected handler
        private void Disconnect(Socket socket)
        {
            var endPoint = (IPEndPoint)socket.RemoteEndPoint;

            lock (connectedSocketsSyncHandle)
            {
                connectedSockets.Remove(endPoint);
            }

            socket.Close();

            OnSocketDisconnected(endPoint);
        }

        private void ReceiveData(byte[] data, IPEndPoint endPoint)
        {
            OnDataReceived(data, endPoint);
        }

        private void EndAcceptSocket(IAsyncResult asyncResult)
        {
            var lister = (TcpListener)asyncResult.AsyncState;
            theLog.Log.Debug("TcpServer.EndAcceptSocket");
            if (disposed)
            {
                theLog.Log.Debug("TcpServer.EndAcceptSocket: tcp server already disposed!");
                return;
            }

            try
            {
                Socket sock;
                try
                {
                    sock = lister.EndAcceptSocket(asyncResult);
                    theLog.Log.DebugFormat("TcpServer.EndAcceptSocket: remote end point: {0}", sock.RemoteEndPoint);
                    Connected(sock);
                }
                finally
                {
                    //EndAcceptSocket can failes, but in any case we want to accept 
new connections
                    lister.BeginAcceptSocket(EndAcceptSocket, lister);
                }

                //we can use this only from .net framework 2.0 SP1 and higher
                var e = new SocketAsyncEventArgs();
                e.Completed += ReceiveCompleted;
                e.SetBuffer(new byte[SocketBufferSize], 0, SocketBufferSize);
                BeginReceiveAsync(sock, e);

            }
            catch (SocketException ex)
            {
                theLog.Log.Error("TcpServer.EndAcceptSocket: failes!", ex);
            }
            catch (Exception ex)
            {
                theLog.Log.Error("TcpServer.EndAcceptSocket: failes!", ex);
            }
        }

        private void BeginReceiveAsync(Socket sock, SocketAsyncEventArgs e)
        {
            if (!sock.ReceiveAsync(e))
            {//IO operation finished syncronously
                //handle received data
                ReceiveCompleted(sock, e);
            }//IO operation finished syncronously
        }

        void ReceiveCompleted(object sender, SocketAsyncEventArgs e)
        {
            var sock = (Socket)sender;
            if (!sock.Connected)
                Disconnect(sock);
            try
            {

                int size = e.BytesTransferred;
                if (size == 0)
                {
                    //this implementation based on IO Completion ports, and in this case
                    //receiving zero bytes mean socket disconnection
                    Disconnect(sock);
                }
                else
                {
                    var buf = new byte[size];
                    Array.Copy(e.Buffer, buf, size);
                    ReceiveData(buf, (IPEndPoint)sock.RemoteEndPoint);
                    BeginReceiveAsync(sock, e);
                }
            }
            catch (SocketException ex)
            {
                //We can't truly handle this excpetion here, but unhandled
                //exception caused process termination.
                //You can add new event to notify observer
                theLog.Log.Error("TcpServer: receive data error!", ex);
            }
            catch (Exception ex)
            {
                theLog.Log.Error("TcpServer: receive data error!", ex);
            }
        }

        private void DisposeImpl(bool manualDispose)
        {
            if (manualDispose)
            {
                //We should manually close all connected sockets
                Exception error = null;
                try
                {
                    if (tcpServer != null)
                    {
                        disposed = true;
                        tcpServer.Stop();
                    }
                }
                catch (Exception ex)
                {
                    theLog.Log.Error("TcpServer: tcpServer.Stop() failes!", ex);
                    error = ex;
                }

                try
                {
                    foreach (var sock in connectedSockets.Values)
                    {
                        sock.Close();
                    }
                }
                catch (SocketException ex)
                {
                    //During one socket disconnected we can faced exception
                    theLog.Log.Error("TcpServer: close accepted socket failes!", ex);
                    error = ex;
                }
                if ( error != null )
                    throw error;
            }
        }


        private void OnSocketConnected(IPEndPoint ipEndPoint)
        {
            var handler = SocketConnected;
            if (handler != null)
                handler(this, new IPEndPointEventArgs(ipEndPoint));
        }

        private void OnSocketDisconnected(IPEndPoint ipEndPoint)
        {
            var handler = SocketDisconnected;
            if (handler != null)
                handler(this, new IPEndPointEventArgs(ipEndPoint));
        }
        private void OnDataReceived(byte[] data, IPEndPoint ipEndPoint)
        {
            var handler = DataReceived;
            if ( handler != null )
                handler(this, new DataReceivedEventArgs(data, ipEndPoint));
        }

        #endregion Private Functions

        //----------------------------------------------------------------------
        //Private Fields
        //----------------------------------------------------------------------
        #region Private Fields
        private const int SocketBufferSize = 1024;
        private readonly TcpListener tcpServer;
        private bool disposed;
        private readonly Dictionary<IPEndPoint, Socket> connectedSockets;
        private readonly object connectedSocketsSyncHandle = new object();
        #endregion Private Fields
    }
}

- Sergey Teplyakov

---

谢谢Sergey，我正在阅读您的源代码以便更好地理解它。我是那种喜欢深入了解事情而不仅仅是复制粘贴的人。 - Martin Milan

我认为这是我要采取的方法 - 使用IO完成端口。感谢您提供完整的代码，但我不会简单地复制和粘贴。正如我之前所说，我喜欢了解为什么某些东西能够工作，因此我将使用您的代码作为参考来实现自己的服务器类。干杯！ - Martin Milan

关于Winsocks API（以及使用IO完成端口），我推荐阅读以下书籍获取更多信息：“Microsoft Windows网络编程”，第二版（这是有关Windows网络编程的最佳书籍之一），以及Jeffrey Richter的“通过C / C ++理解Windows”（这是有关Win32，多线程和IO完成端口的最佳书籍之一）。很高兴我能够帮助！ - Sergey Teplyakov

Answer 2

制作多线程服务器非常简单。看看这个例子。

class Server
{
    private Socket socket;
    private List<Socket> connections;
    private volatile Boolean endAccept;

    // glossing over some code.


    /// <summary></summary>
    public void Accept()
    {
        EventHandler<SocketAsyncEventArgs> completed = null;
        SocketAsyncEventArgs args = null;

        completed = new EventHandler<SocketAsyncEventArgs>((s, e) =>
        {
            if (e.SocketError != SocketError.Success)
            {
                // handle
            }
            else
            {
                connections.Add(e.AcceptSocket);
                ThreadPool.QueueUserWorkItem(AcceptNewClient, e.AcceptSocket);
            }

            e.AcceptSocket = null;
            if (endAccept)
            {
                args.Dispose();
            }
            else if (!socket.AcceptAsync(args))
            {
                completed(socket, args);
            }
        });

        args = new SocketAsyncEventArgs();
        args.Completed += completed;

        if (!socket.AcceptAsync(args))
        {
            completed(socket, args);
        }
    }

    public void AcceptNewClient(Object state)
    {
        var socket = (Socket)state;
        // proccess        
    }        
}

- ChaosPandion

---

谢谢你的回复，ChoasPandion...不幸的是我现在要出门了，但稍后我会看一下这个。Lambda函数？语法还在摸索中... - Martin Milan

我们可以使用BeginAcceptSocket。 - Sergey Teplyakov

使用AcceptAsync有明显的优势。当使用BeginAccept时，需要每次创建一个新的IAsyncResult对象。 对于高性能服务器，您希望尽可能减少对象的创建。使用此方法，您可以创建一个对象并在服务器的持续时间内重复利用它。 - ChaosPandion

Choas/Sergey，感谢您的评论。然而，这种对象创建开销可能有多大呢？如果像Sergey所建议的那样简单，避免处理线程可能会更简单... - Martin Milan

忽略我的评论，开销对你来说相关性的可能性很小。实际上，你没有理由不使用线程。最棒的是，每次使用线程它们都会变得越来越容易。 - ChaosPandion

Answer 3

来自主要处理移动网络的人的一些建议：在本地主机上使用常规网络连接做好功课，这将节省您在测试期间的大量时间，并使您保持理智，直到找到最适合您的方法。

至于某些特定的实现，我总是使用同步套接字（您需要配置超时以避免出现问题时被卡住），并且所有内容都在单独的线程中运行，这些线程通过事件进行同步。它比你想象的要简单得多。以下是一些有用的链接，可帮助您入门：

• http://msdn.microsoft.com/en-us/library/3e8s7xdd.aspx
• http://msdn.microsoft.com/en-us/library/ms228969.aspx

- wasker

---

干杯 - 我会把这加到需要查看的事项列表里...谁能想到这会这么有趣呢？ - Martin Milan

Answer 4

我现在正在编写同样的应用程序，我使用类似这样的解决方案：

http://clutch-inc.com/blog/?p=4

现在已经进行了测试，而且完美地运作。重要的是将此服务仅用于接收和存储消息（某个地方），而不进行其他工作。我正在使用 NServiceBus 来保存消息。其他服务从队列中获取消息并完成其余部分。

- dariol

---

谢谢Dario - 我也会研究一下那个。 - Martin Milan

Answer 5

嗯，C#语法现在不是很熟悉，但我认为它与Posix标准并没有太大的区别。

你可以在创建监听套接字时指定backlog的值（该服务器的最大同时连接数），并创建一个相同大小的线程池。线程池比传统的更易于使用。TCP会将超过backlog参数的所有连接排队给你。

- Andres