在C#中的异步套接字服务器,通过套接字服务器实现客户端之间的通信。

3
我正在尝试使用c#开发服务器/客户端异步套接字。我已经按照MSDN链接上的指南进行操作。在我的情况下,套接字服务器监听特定端点,许多客户端可以同时连接到服务器,客户端可以与服务器通信,服务器可以与客户端通信。假设客户端1和客户端2连接了服务器,客户端1可以向服务器发送消息,服务器可以发送给客户端1,客户端2也是如此。现在,我希望客户端能够通过服务器彼此通信。例如;客户端2想要与客户端1通信,为此,客户端2将向服务器发送消息(此消息将包含一些预设字符),然后服务器将从客户端2接收文本,并获取客户端1的处理程序并将此消息发送给客户端1,客户端1将响应服务器。现在我想将客户端1对该消息的响应发送给客户端2,但我不知道该如何做,因为客户端1通过自己的处理程序与服务器通信,我被卡住了,非常感谢您的帮助! 我的代码如下:
using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading;
using System.Threading.Tasks;
using System.Net;
using System.Net.Sockets;



namespace SocketServer
{
    // State object for reading client data asynchronously  
    public class StateObject
    {
        // Client  socket.  
        public Socket workSocket = null;
        // Size of receive buffer.  
        public const int BufferSize = 1024;
        // Receive buffer.  
        public byte[] buffer = new byte[BufferSize];
        // Received data string.  
        public StringBuilder sb = new StringBuilder();

        public int clientNumber;
    }
    public class AsyncSocketServer
    {
        public static ManualResetEvent allDone = new ManualResetEvent(false);

        public static Dictionary<int, StateObject> Clients = new Dictionary<int, StateObject>();

        public static int connectedClient = 0;




        public AsyncSocketServer()
        {


        } 
        public static void startListening() {

            Byte[] bytes = new Byte[1024];
            int Port = 1122;

            IPAddress IP = IPAddress.Parse("127.0.0.1");
            IPEndPoint EP = new IPEndPoint(IP, Port);
            Socket listner = new Socket(IP.AddressFamily, SocketType.Stream, ProtocolType.Tcp);



            try
            {
                listner.Bind(EP);
                listner.Listen(100);

                while (true)
                {
                   allDone.Reset();

                    Console.WriteLine("Waiting for the Connection......");

                    listner.BeginAccept(new AsyncCallback(AcceptCallBack), listner);

                    allDone.WaitOne();
                }


            }
            catch(Exception e)
            {
                Console.WriteLine("Exception Occured ! in start listening method "+e.ToString());
            }

             Console.WriteLine("\nPress ENTER to continue...");  
            Console.Read();  

        }

        public static void AcceptCallBack(IAsyncResult ar)
        {
            connectedClient++;
            Console.WriteLine("client number " + connectedClient);
            allDone.Set();



            Socket listner = (Socket)  ar.AsyncState;
            Socket handler = listner.EndAccept(ar);

            StateObject state = new StateObject();
            state.clientNumber = connectedClient;

            Clients.Add(connectedClient, state);
           Console.WriteLine("total clients {0}",Clients.Count());

            state.workSocket = handler;
            handler.BeginReceive(state.buffer, 0, StateObject.BufferSize,0,new AsyncCallback(ReadCallBack),state);

        }
        public static void ReadCallBack(IAsyncResult ar)
        {  

        String content = String.Empty;



        // Retrieve the state object and the handler socket  
        // from the asynchronous state object.  
        try { 
        StateObject state = (StateObject) ar.AsyncState;
        state.sb.Clear();
        Socket handler = state.workSocket;  

        // Read data from the client socket.   
        int bytesRead = handler.EndReceive(ar);  

        if (bytesRead > 0) {  
            // There  might be more data, so store the data received so far.  
            state.sb.Append(Encoding.ASCII.GetString(  
                state.buffer,0,bytesRead));  

            // Check for end-of-file tag. If it is not there, read   
            // more data.  

            content = state.sb.ToString();

            if (content.Substring(0, 3) == "cmd") {
                foreach (StateObject Client in Clients.Values) {
                    if (Client.clientNumber == 1) { 
                        Console.WriteLine("value is "+Client.clientNumber);
                        if (isClientConnected(Client.workSocket)){
                            Send(Client.workSocket, "did you receive my message");
                            //now client number 1 will response through its own handler, but i want to get response of 
                            //client number 1 and return this response to client number 2

                        }
                        else {
                            string responsemsg = "client number " + Client.clientNumber + " is disconnected !";
                            Console.WriteLine(responsemsg);
                            Send(handler,responsemsg);
                        }
                    }

                }
            }

            Console.WriteLine("Read {0} bytes from client {1} socket. \n Data : {2}",
                    content.Length, state.clientNumber,content);
            // Echo the data back to the client.  

            if (isClientConnected(handler))
            {
                Send(handler, content);
            }
            handler.BeginReceive(state.buffer, 0, StateObject.BufferSize, 0, new AsyncCallback(ReadCallBack), state);

        }
        }
        catch (SocketException e)
        {
            //once if any client disconnected then control will come into this block
            Console.WriteLine("Socket Exception Occured in Read Call Back : " + e.Message.ToString());

        }
        catch (Exception e)
        {
            //once if any client disconnected then control will come into this block
            Console.WriteLine("Exception Occured in Read Call Back : " + e.Message.ToString());

        }
        }
        private static void Send(Socket handler, String data)
        {
            // Convert the string data to byte data using ASCII encoding.  
            byte[] byteData = Encoding.ASCII.GetBytes(data);

            // Begin sending the data to the remote device.  
            handler.BeginSend(byteData, 0, byteData.Length, 0,
                new AsyncCallback(SendCallback), handler);
        }

        private static void SendCallback(IAsyncResult ar)
        {
            try
            {
                // Retrieve the socket from the state object.  
                Socket handler = (Socket)ar.AsyncState;

                // Complete sending the data to the remote device.  
                int bytesSent = handler.EndSend(ar);
                Console.WriteLine("Sent {0} bytes to client.", bytesSent);



                //handler.Shutdown(SocketShutdown.Both);
                //handler.Close();

            }
            catch (Exception e)
            {
                Console.WriteLine(e.ToString());
            }
        }

        public static bool isClientConnected(Socket handler){

            return handler.Connected;
        }
        public static int Main(string[] args)
        {

            startListening();


            return 0;


        }
    }
}

针对未来的读者:.Net Core团队建议不要使用此问题中的“旧”链接,而是使用带有Async后缀的方法。https://github.com/dotnet/core/issues/4828#issuecomment-643619106 - Abraham
2个回答

1
对于任何复杂的基于套接字的应用程序,我建议使用套接字库(如DotNetty)来抽象传输层,使您可以专注于应用程序逻辑。请查看他们的SecureChat示例,它可能与您尝试实现的内容非常相似。
我提供了一个快速示例,使用DotNetty服务器允许客户端通过向服务器注册并在客户端之间路由消息来发送命令。
using System;
using System.Collections.Concurrent;
using System.Threading;
using System.Threading.Tasks;
using DotNetty.Transport.Channels;
using Newtonsoft.Json;
using System.IO;

namespace MultiClientSocketExample
{
    public enum Command
    {
        Register = 1,  // Register a new client
        SendToClient = 2, // Send a message from one client to antoher
        DoClientAction = 3 // Replace this with your client-to-client command
    }

    // Envelope for all messages handled by the server
    public class Message
    {
        public string ClientId { get; set; }
        public Command Command { get; set; }
        public string Data { get; set; }
    }

    // Command for seding a message from one client to antoher.   
    // This would be serialized as JSON and stored in the 'Data' member of the Message object.
    public class SendToClientCommand
    {
        public string DestinationClientId { get; set; }  // The client to receive the message

        public Command ClientCommand { get; set; } // The command for the destination client to execute

        public string Data { get; set; } // The payload for the destination client
    }

    // An object for storing unhandled messages in a queue to be processed asynchronously
    // This allows us to process messages and respond to the appropriate client,
    // without having to do everything in the ChannelRead0 method and block the main thread
    public class UnhandledMessage
    {
        private readonly Message message;
        private readonly IChannelHandlerContext context;

        public UnhandledMessage(Message message, IChannelHandlerContext context)
        {
            this.message = message;
            this.context = context;
        }

        public Message Message => message;
        public IChannelHandlerContext Context => context;

        public Command Command => message.Command;
        public string ClientId => message.ClientId;
        public string Data => message.Data;
    }

    // A representation of the connected Clients on the server.  
    // Note:  This is not the 'Client' class that would be used to communicate with the server.
    public class Client
    {
        private readonly string clientId;
        private readonly IChannelHandlerContext context;

        public Client(string clientId, IChannelHandlerContext context)
        {
            this.clientId = clientId;
            this.context = context;
        }

        public string ClientId => clientId;
        public IChannelHandlerContext Context => context;
    }

    // The socket server, using DotNetty's SimpleChannelInboundHandler
    // The ChannelRead0 method is called for each Message received
    public class Server : SimpleChannelInboundHandler<Message>, IDisposable
    {
        private readonly ConcurrentDictionary<string, Client> clients;
        private readonly ConcurrentQueue<UnhandledMessage> unhandledMessages;
        private readonly CancellationTokenSource cancellation;
        private readonly AutoResetEvent newMessage;

        public Server(CancellationToken cancellation)
        {
            this.clients = new ConcurrentDictionary<string, Client>();
            this.newMessage = new AutoResetEvent(false);
            this.cancellation = CancellationTokenSource.CreateLinkedTokenSource(cancellation);
        }

        // The start method should be called when the server is bound to a port.
        // Messages will be received, but will not be processed unless/until the Start method is called
        public Task Start()
        {
            // Start a dedicated thread to process messages so that the ChannelRead operation does not block
            return Task.Run(() =>
            {
                var serializer = JsonSerializer.CreateDefault();  // This will be used to deserialize the Data member of the messages

                while (!cancellation.IsCancellationRequested)
                {
                    UnhandledMessage message;
                    var messageEnqueued = newMessage.WaitOne(100);  // Sleep until a new message arrives

                    while (unhandledMessages.TryDequeue(out message))  // Process each message in the queue, then sleep until new messages arrive
                    {
                        if (message != null)
                        {
                            // Note: This part could be sent to the thread pool if you want to process messages in parallel
                            switch (message.Command)
                            {
                                case Command.Register:
                                    // Register a new client, or update an existing client with a new Context
                                    var client = new Client(message.ClientId, message.Context);
                                    clients.AddOrUpdate(message.ClientId, client, (_,__) => client);
                                    break;
                                case Command.SendToClient:
                                    Client destinationClient;
                                    using (var reader = new JsonTextReader(new StringReader(message.Data)))
                                    {
                                        var sendToClientCommand = serializer.Deserialize<SendToClientCommand>(reader);
                                        if (clients.TryGetValue(sendToClientCommand.DestinationClientId, out destinationClient))
                                        {
                                            var clientMessage = new Message { ClientId = message.ClientId, Command = sendToClientCommand.ClientCommand, Data = sendToClientCommand.Data };
                                            destinationClient.Context.Channel.WriteAndFlushAsync(clientMessage);
                                        }
                                    }
                                    break;
                            }
                        }
                    }
                }
            }, cancellation.Token);
        }

        // Receive each message from the clients and enqueue them to be procesed by the dedicated thread
        protected override void ChannelRead0(IChannelHandlerContext context, Message message)
        {
            unhandledMessages.Enqueue(new UnhandledMessage(message, context));
            newMessage.Set(); // Trigger an event so that the thread processing messages wakes up when a new message arrives
        }

        // Flush the channel once the Read operation has completed
        public override void ChannelReadComplete(IChannelHandlerContext context)
        {
            context.Flush();
            base.ChannelReadComplete(context);
        }

        // Automatically stop the message-processing thread when this object is disposed
        public void Dispose()
        {
            cancellation.Cancel();
        }
    }
}

1
TcpListener 有什么问题吗?至少它支持 async/await。相比于为本质上是 I/O 绑定的事情启动 CPU 绑定线程,这样做更好。 - user585968
谢谢提供的链接。我看了一下你提到的SecureChat示例,它似乎更像是一个高级EAI库(通常是传输无关的),而不是TCP套接字库。这并没有什么问题,但这不是苹果和苹果之间的比较。无论如何,你的示例可能有效,但它类似于旧式的重叠I/O技术,即生成一个线程并在句柄上睡眠。这不是使用“任务”的推荐方法,而且浪费了一个本来可以在后台进行IOCP的线程,而其他事情则可以更好地完成。祝你好运+1。 - user585968
我尝试了DotNetty,并且专注于代码时,我注意到该套接字服务器向所有连接的客户端广播消息,而要将库更改为仅发送消息给特定客户端需要很多经验。请问现在我应该怎么做,才能修改库的默认行为?能否提供建议? - Kashif Ali
你能上传你的代码(或者一个等效的重现行为的示例)吗?这样我们就可以看看了。 - Aaron M. Eshbach
#Aaron M.Eshbach,我已经编辑了问题部分,包括代码和我的评论,请仔细阅读。非常感谢您的帮助。 - Kashif Ali
显示剩余3条评论

0

我尝试过相同的方法,也是基于 MSDN 相同的代码。 可能的解决方案是使用一个套接字列表:

List<Socket> clients = new List<Socket>();

然后,当客户端连接时,将客户端添加到列表中。
public void AcceptCallback(IAsyncResult ar)
    {
        // Signal the main thread to continue.  
        allDone.Set();

        // Get the socket that handles the client request.  
        Socket listener = (Socket)ar.AsyncState;
        Socket handler = listener.EndAccept(ar);

        clients.Add(handler);
        ...
    }

你需要知道每个连接的客户端的ID(句柄),然后才能向特定的客户端发送一些消息:

 public void SendToOne(string id,string message)
    {
        foreach (Socket s in clients)
        {
            if (s.Handle.ToString() == id)
            {
                Send(s, message);
            }
        }

    }

网页内容由stack overflow 提供, 点击上面的
可以查看英文原文,
原文链接