c# - C#中的异步套接字服务器，客户端通过套接字服务器进行客户端通信

Question

我正在尝试使用 C# 开发服务器/客户端异步套接字。我遵循了 MSDN 上的指南关联。在我的例子中，套接字服务器监听特定端点，许多客户端可以一次连接到服务器，客户端可以与服务器通信，服务器可以与客户端通信。假设客户端 1 和客户端 2 与服务器连接，客户端 1 可以向服务器发送消息，服务器可以向客户端 1 发送消息，客户端 2 的情况相同。现在我希望客户端应该能够通过服务器相互通信。例如:client 2 想和client 1 通信，因为client 2 会向server 发送消息(这个消息会包含一些预设的字符；)，然后server 会收到client 2 发来的文本，并得到client 1 的handler 并发送此消息发送给客户端 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;


        }
    }
}

Answer 1

对于任何复杂的基于套接字的应用程序，我建议使用类似 DotNetty 的套接字库抽象传输层并让您专注于您的应用程序逻辑。查看他们的 SecureChat example ，它可能与您要实现的目标非常相似。

我整理了一个 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();
        }
    }
}