сервер отправка сообщения всем подключенным клиентам

Question

я знаю, этот вопрос был задан раньше, и я попробовал различные решения, но я застрял в части реализации .. :(

В настоящее время несколько клиентов могут подключаться к сервер, я использовал многопотоковый пример сервера/клиента KnockKnock из javadocs и немного отредактировал его, чтобы вы могли просто отправлять сообщения на сервер, и он будет отсылать их обратно вам, но я хочу, чтобы это можно было сделать так, чтобы если клиент 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); 
     } 
    } 



} 

ServerThread

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(); 
     } 
    } 


} 

Client (не уверено, что в случае необходимости, но здесь это так или иначе)

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

Операция «записи» преграждает в вашем примере. Таким образом, повторение всех подключений может привести к задержкам и блокировке push-потока. Также всегда устанавливайте 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); 
    } 
}