服务器向所有连接的客户端发送消息

Question

我知道以前有人问过这个问题，并且我尝试了不同的解决方案，但我陷入了实现部分.. :(

当前多个客户端可以连接到服务器，我使用了 javadocs 中的多线程 KnockKnock 服务器/客户端示例，并对其稍作编辑，以便您可以将消息发送到服务器，它会将它们回显给您，但我希望能够做到这一点，如果客户端 1 发送消息，那么服务器会将它们广播回连接到服务器的所有客户端。

我试着环顾四周，看到和我现在处于相同位置的人，他们被告知要制作一个列表来跟踪所有连接，然后遍历列表并发送消息，但是我真的不知道该放在哪个班级，也不知道如何处理。

如果有人可以告诉我或者只是给我提示我应该从哪里开始，我将不胜感激，因为我现在真的只是卡住了:(

这是我目前所处的位置：

服务器：

    import java.io.IOException;
import java.net.ServerSocket;
import java.net.Socket;
import java.util.ArrayList;
import java.util.List;

public class Server {



    public static void main(String[] args) throws IOException {



        boolean listening = true;

        try (ServerSocket serverSocket = new ServerSocket(4444)) {

            while (listening) {
                ServerThread thread = new ServerThread(serverSocket.accept());
                thread.start();
            }
        } catch (IOException e) {
            System.err.println("Could not listen on port " );
            System.exit(-1);
        }
    }



}

服务器线程

    import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.io.PrintWriter;
import java.net.Socket;
import java.util.ArrayList;
import java.util.List;


public class ServerThread extends Thread{



    private Socket socket = null;



    public ServerThread(Socket socket) {
        super("MultiServerThread");
        this.socket = socket;

    }

    public void run() {

        try (
                PrintWriter out = new PrintWriter(socket.getOutputStream(), true);
                BufferedReader in = new BufferedReader(
                        new InputStreamReader(
                                socket.getInputStream()));
        ) {

            while (true) {


                String input = in.readLine();
                System.out.println(input);
                out.println("ecco " + input);



                if (input.equals("Bye"))
                    break;
            }
            socket.close();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }


}

客户（不确定是否有必要，但无论如何都是这样）

    import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.io.PrintWriter;
import java.net.Socket;
import java.net.UnknownHostException;


public class Client {

    public static void main(String[] args) throws IOException {


        try (
                Socket kkSocket = new Socket("172.30.242.51", 4444);
                PrintWriter out = new PrintWriter(kkSocket.getOutputStream(), true);
                BufferedReader in = new BufferedReader(
                        new InputStreamReader(kkSocket.getInputStream()));
        ) {
            BufferedReader stdIn =
                    new BufferedReader(new InputStreamReader(System.in));

            while (true) {

                if(in != null) {

                    String input = stdIn.readLine();
                    out.println("Client: " + input);
                    System.out.println(in.readLine());
                    out.flush();

                }

            }
        } catch (UnknownHostException e) {
            System.err.println("Don't know about host " );
            System.exit(1);
        } catch (IOException e) {
            System.err.println("Couldn't get I/O for the connection to " );
            System.exit(1);
        }
    }
}

周末愉快=）

Answer 1

在您的示例中，“写入”操作被阻止。因此，通过所有连接进行迭代可能会导致延迟并阻塞您的推送线程。如果您不想发生内存泄漏，也请始终为套接字设置 SO_TIMEOUT。

我建议使用netty server

它具有非常好的功能，可以将数据推送到所有连接的客户端 - 查找 ChannelGroup

Answer 2

为什么不使用 NIO 来解决这个问题？

一个简单的例子：

public class EchoServer {

  public static void main(String[] args) throws Exception {
    //Create TCP server channel
    ServerSocketChannel serv = ServerSocketChannel.open();
    ServerSocket sock = serv.socket();

    //Create a socket on your IP and port (i.e: localhost:12345)
    SocketAddress addr = new InetSocketAddress(12345);

    //Bind server socket and socket address
    sock.bind(addr);

    //Configure socket so all its methods won't be blocking
    serv.configureBlocking(false);

    //Create a selector to attend all the incoming requests
    Selector selector = Selector.open();

    //Register into the selector the accept request type
    serv.register(selector,SelectionKey.OP_ACCEPT);

    //Create a common buffer
    ByteBuffer commonBuffer = ByteBuffer.allocate(10000);
    commonBuffer.clear();

    Iterator<SelectionKey> it = null;
    ByteBuffer channelBuffer  = null;
    for (;;){ //Infinite loop

        System.out.println("Waiting for events......");
        selector.select(); // This call do is blocking

        System.out.println("New event received");

        it = selector.selectedKeys().iterator();

        while(it.hasNext()) {
            SelectionKey key = (SelectionKey) it.next();
            System.out.println(String.format("Processing %s", key));
            it.remove(); // Remove it to avoid duplications

            try{
                if (key.isAcceptable()) {
                    System.out.println("Received new connection request");
                    processConnectionRequest(serv, selector);
                }else if (key.isReadable()) {
                    System.out.println("Received new reading request");
                    processReadingRequest(selector, commonBuffer, key);
                }else if (key.isWritable()) {
                    System.out.println("Received new writing request");
                    processWritingRequest(key);
                }
            }catch(Exception e){
                key.cancel();
                try {
                    key.channel().close();
                } catch (Exception ce) {}
            }//end catch
        }//end while
    }//end for
  }//end main

  private static void processWritingRequest(SelectionKey key) throws IOException {
    SocketChannel cli = (SocketChannel) key.channel();
    ByteBuffer buf = (ByteBuffer) key.attachment();

    System.out.println(String.format("Wrinting into the channel %s", cli));
    buf.flip();//prepare the buffer
    buf.rewind();
    cli.write(buf);

    if (buf.hasRemaining()) {
        //If there is more content remaining, compact the buffer    
        buf.compact();
    } else {
        buf.clear();
        key.interestOps(SelectionKey.OP_READ);
    }
  }

  private static void processReadingRequest(Selector selector, ByteBuffer commonBuffer, SelectionKey key)
        throws IOException {
    SocketChannel cli = (SocketChannel) key.channel();


    if (cli.read(commonBuffer) == -1) {
        System.out.println(String.format("Closing channel %s", cli));
        cli.close(); // internally calls key.cancel()
    }
    else {//Send the data to all the channels

        commonBuffer.flip();//prepare the buffer
        Iterator<SelectionKey> it2 = selector.keys().iterator();
        System.out.println("Writing data to all the channels");
        SelectionKey keys = null;
        while(it2.hasNext()) {
            keys = it2.next();
            System.out.println(String.format("Writing in %s", keys));

            ByteBuffer buf = (ByteBuffer) keys.attachment();

            if(buf!=null)
            {
            buf.put(commonBuffer);
            keys.interestOps(SelectionKey.OP_WRITE|SelectionKey.OP_READ);

            commonBuffer.rewind();
            }

        }
        commonBuffer.clear();

    }
  }

  private static void processConnectionRequest(ServerSocketChannel serv, Selector selector)
        throws IOException, ClosedChannelException {
    ByteBuffer channelBuffer;
    SocketChannel cli = serv.accept();
    cli.configureBlocking(false);
    channelBuffer = ByteBuffer.allocate(10000);
    System.out.println(String.format("Registering new reading channel: %s", cli));
    cli.register(selector, SelectionKey.OP_READ, channelBuffer);
  }
}