在java中将msg从发送者发送到接收者（在并发模式下）

Question

我想编写程序，使用一台服务器从特定端口接收来自记者的消息，并将在线消息发送到连接在另一个特定端口上的客户端（阅读器）......

提示1 => 我的服务器有两个开放端口...一个专用于发件人，另一个用于阅读器...

提示2 => 当从服务器中的发件人接收到消息时 => 我们（服务器）应该将此消息发送给所有监听的在线客户端

我该如何实现它们... 我在将消息从服​​务器发送到“所有”客户端时遇到问题！

Answer 1

你可以这样做 -

public class SenderReceiver {

    private static volatile boolean serverStarted;
    private static final String SERVER_HOST = "127.0.0.1";
    private static final int SERVER_PORT = 25000;
    private static final List<ReceiverConnectionHandler> handlers = new CopyOnWriteArrayList<ReceiverConnectionHandler>();
    private static final ExecutorService executorService = Executors.newCachedThreadPool();

    public static void main(String[] args) {
        // start server
        executorService.submit(new SenderReceiver().new Server());
        // wait until started
        while (!serverStarted) {
            try {
                Thread.sleep(100);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
        // start receiver threads
        int i = 0;
        while (i++ < 10) {
            executorService.submit(new Receiver(SERVER_HOST, SERVER_PORT));
        }
        // start sender threads
        i = 0;
        while (i++ < 5) {
            executorService.submit(new Sender(SERVER_HOST, SERVER_PORT));
        }
    }

    private class Server implements Runnable {
        public void run() {
            try {
                ServerSocket serverSocket = new ServerSocket(SERVER_PORT);
                serverStarted = true;
                System.out.println("Server started");
                while (true) {
                    Socket socket = serverSocket.accept();
                    BufferedReader br = new BufferedReader(new InputStreamReader(socket.getInputStream()));
                    String connectionName = br.readLine();
                    if ("sender".equals(connectionName)) {
                        executorService.submit(new SenderConnectionHandler(br, handlers));
                    } else {
                        ReceiverConnectionHandler handler = new ReceiverConnectionHandler(socket);
                        handlers.add(handler);
                        executorService.submit(handler);
                    }
                }
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }

    private class SenderConnectionHandler implements Runnable {
        private BufferedReader br;
        private List<ReceiverConnectionHandler> handlers;

        public SenderConnectionHandler(BufferedReader br, List<ReceiverConnectionHandler> handlers) {
            this.br = br;
            this.handlers = handlers;
        }

        public void run() {
            try {
                String message;
                while ((message = br.readLine()) != null) {
                    Thread.sleep(1000); // just allowing all clients to be registered
                    for (ReceiverConnectionHandler handler : handlers) {
                        handler.update(message);
                    }
                }
            } catch (Exception e) {
                e.printStackTrace();
            } finally {
                if (br != null) {
                    try {
                        br.close();
                    } catch (Exception e) {
                        e.printStackTrace();
                    }
                }
            }

        }
    }

    private class ReceiverConnectionHandler implements Runnable {
        private Socket receiverSocket;
        private final BlockingQueue<String> messages = new LinkedBlockingQueue<>();

        public ReceiverConnectionHandler(Socket socket) {
            this.receiverSocket = socket;
        }

        public void run() {
            BufferedWriter bw = null;
            try {
                bw = new BufferedWriter(new OutputStreamWriter(receiverSocket.getOutputStream()));
                String message;
                while ((message = messages.poll(1, TimeUnit.SECONDS)) != null) {
                    bw.write(message);
                    bw.newLine();
                    bw.flush();
                }
            } catch (Exception e) {
                e.printStackTrace();
            } finally {
                if (bw != null) {
                    try {
                        bw.close();
                    } catch (Exception e) {
                        e.printStackTrace();
                    }
                }
            }

        }

        public void update(Object message) {
            if (message instanceof String) {
                messages.add((String) message);
            }
        }
    }
}

class Sender implements Runnable {
    private static AtomicInteger senderCount = new AtomicInteger(0);
    private String serverHost;
    private int serverPort;

    public Sender(String serverHost, int serverPort) {
        this.serverHost = serverHost;
        this.serverPort = serverPort;
    }

    public void run() {
        BufferedWriter bw = null;
        try {
            Socket socket = new Socket(serverHost, serverPort);
            int senderNumber = senderCount.incrementAndGet();
            bw = new BufferedWriter(new OutputStreamWriter(socket.getOutputStream()));
            bw.write("sender");
            bw.newLine();
            String message = "This is a test message from sender " + senderNumber;
            bw.write(message);
            bw.newLine();
            System.out.println("Sender " + senderNumber + " : " + message);
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            if (bw != null) {
                try {
                    bw.close();
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        }
    }
}

class Receiver implements Runnable {
    private static AtomicInteger receiverCount = new AtomicInteger(0);
    private String serverHost;
    private int serverPort;

    public Receiver(String serverHost, int serverPort) {
        this.serverHost = serverHost;
        this.serverPort = serverPort;
    }

    public void run() {
        BufferedReader br = null;
        BufferedWriter bw = null;
        try {
            Socket socket = new Socket(serverHost, serverPort);
            int receiverNumber = receiverCount.incrementAndGet();
            bw = new BufferedWriter(new OutputStreamWriter(socket.getOutputStream()));
            bw.write("client");
            bw.newLine();
            bw.flush();
            br = new BufferedReader(new InputStreamReader(socket.getInputStream()));
            String message;
            while ((message = br.readLine()) != null) {
                System.out.println("Receiver " + receiverNumber + ": " + message);
            }
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            if (br != null) {
                try {
                    br.close();
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
            if (bw != null) {
                try {
                    bw.close();
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        }
    }
}