Socket结合线程池使用实现客户端和服务端通信demo

首先，我们需要先了解 Socket 是什么。 Socket 是一种网络通信协议，它能够在计算机之间实现双向通信。在使用 Socket 进行通信时，通常需要使用线程池，以便能够同时处理多个连接。

接下来，我们将演示如何使用 Socket 和线程池来实现一个基本的客户端和服务端通信 Demo，包含两个示例：

示例一：实现一个简单的客户端和服务端通信

首先，我们需要启动服务端程序，代码如下：

import java.net.ServerSocket;
import java.net.Socket;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class Server {
    private static final int PORT = 8080; // 服务端监听的端口号
    private static final int THREADS_NUM = 10; // 线程池中线程的数量

    public static void main(String[] args) {
        try (ServerSocket serverSocket = new ServerSocket(PORT)) {
            ExecutorService pool = Executors.newFixedThreadPool(THREADS_NUM); // 创建线程池
            System.out.println("Server is running...");
            while (true) {
                Socket socket = serverSocket.accept(); // 等待客户端连接
                pool.execute(new ServerWorker(socket)); // 把客户端连接交给线程池处理
            }
        } catch (Exception e) {
            e.printStackTrace();
            System.exit(1);
        }
    }
}

在上面的代码中，我们创建了一个 ServerSocket 并监听指定的端口号 PORT。当有客户端连接到该端口时，创建一个 Socket 并将其传递给 ServerWorker 对象，以便线程池处理。

下面是 ServerWorker 的代码：

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

public class ServerWorker implements Runnable {
    private final Socket socket;

    public ServerWorker(Socket socket) {
        this.socket = socket;
    }

    @Override
    public void run() {
        try (BufferedReader in = new BufferedReader(new InputStreamReader(socket.getInputStream()));
             PrintWriter out = new PrintWriter(socket.getOutputStream(), true)) {
            String message;
            while ((message = in.readLine()) != null) {
                System.out.println("Server received message: " + message);
                out.println("Server response: " + message.toUpperCase()); // 将请求转化为大写并响应
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}

ServerWorker 实现了 Runnable 接口，用于执行线程池中的任务。在 run 方法中，我们通过 InputStream 和 OutputStream 对客户端传递的请求进行处理，并通过 OutputStream 将响应传递给客户端。

现在，我们可以启动客户端程序，并向服务端发送请求，代码如下：

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

public class Client {
    private static final String HOST = "127.0.0.1"; // 服务端 IP 地址
    private static final int PORT = 8080; // 服务端端口号

    public static void main(String[] args) {
        try (Socket socket = new Socket(HOST, PORT);
             BufferedReader in = new BufferedReader(new InputStreamReader(socket.getInputStream()));
             PrintWriter out = new PrintWriter(socket.getOutputStream(), true)) {
            out.println("Hello, Server!"); // 向服务端发送请求
            String response = in.readLine();
            System.out.println("Client received response: " + response);
        } catch (Exception e) {
            e.printStackTrace();
            System.exit(1);
        }
    }
}

我们可以看到，客户端程序中，我们通过 Socket 向服务端发送请求，并读取服务端响应。

示例二：使用多路复用的方式实现客户端和服务端通信

接下来，我们将演示如何使用多路复用的方式来实现客户端和服务端通信。

在这种实现方式中，我们将使用 ServerSocketChannel 和 SocketChannel，以及 Selector 对象来实现多路复用。具体代码如下：

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.*;
import java.util.Iterator;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.stream.IntStream;

public class MultiplexingServer {
    private static final int PORT = 8080;
    private static final int THREADS_NUM = 10;
    private boolean running = true;

    public void run() throws IOException {
        try (ServerSocketChannel serverChannel = ServerSocketChannel.open()) {
            serverChannel.configureBlocking(false);
            serverChannel.socket().bind(new InetSocketAddress(PORT));

            Selector selector = Selector.open();
            serverChannel.register(selector, SelectionKey.OP_ACCEPT);

            ExecutorService pool = Executors.newFixedThreadPool(THREADS_NUM);

            ConcurrentLinkedQueue<SocketChannel> clientSockets = new ConcurrentLinkedQueue<>();

            IntStream.range(0, THREADS_NUM).forEach(i -> {
                pool.execute(() -> {
                    while (running) {
                        try {
                            selector.select(); // 阻塞
                            Iterator<SelectionKey> keys = selector.selectedKeys().iterator();
                            while (keys.hasNext()) {
                                SelectionKey key = keys.next();
                                if (key.isAcceptable()) {
                                    ServerSocketChannel server = (ServerSocketChannel) key.channel();
                                    SocketChannel client = server.accept();
                                    client.configureBlocking(false);
                                    client.register(selector, SelectionKey.OP_READ);
                                    clientSockets.offer(client);
                                    System.out.println("Client accepted: " + client.getRemoteAddress());
                                } else if (key.isReadable()) {
                                    SocketChannel client = (SocketChannel) key.channel();
                                    ByteBuffer buffer = ByteBuffer.allocate(1024);
                                    client.read(buffer);
                                    buffer.flip();
                                    String message = new String(buffer.array()).trim();
                                    System.out.println("Server received message: " + message);
                                    clientSockets.stream().filter(socketChannel -> client != socketChannel)
                                            .forEach(socketChannel -> {
                                                try {
                                                    socketChannel.write(ByteBuffer.wrap(message.toUpperCase().getBytes()));
                                                } catch (IOException e) {
                                                    e.printStackTrace();
                                                }
                                            });
                                }
                                keys.remove();
                            }
                        } catch (Exception e) {
                            e.printStackTrace();
                        }
                    }
                    System.out.println("Server closed.");
                });
            });

            // Ctrl-C 终止启动的程序
            Runtime.getRuntime().addShutdownHook(new Thread(() -> {
                running = false;
                selector.wakeup();
                System.out.println("Shutting down...");
                pool.shutdown();
            }));

            System.out.println("Server is running...");
            while (true) {
                Thread.sleep(1000);
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}

首先，我们创建一个 ServerSocketChannel，并设置为非阻塞模式。然后，我们将 ServerSocketChannel 注册到 Selector 中，以便能够监听客户端连接请求。

接下来，我们创建一个线程池，并用于处理客户端 Socket 连接请求。当有客户端连接请求到达时，我们创建一个 SocketChannel，并将其注册到 Selector 中。

在客户端发来消息时，我们通过 Selector 获取可读事件，并对其进行处理。对于每个读事件，我们向所有客户端发送相同的消息，这里采用了并发队列来管理多个 SocketChannel。

最后，我们通过启动线程来处理所有客户端请求，并使用 Selector 对客户端进行监听，以便能够接收客户端的请求和响应。

除此之外，我们为应用程序添加了一些优雅地关闭的支持，以便能够在用户按下 Ctrl-C 停止应用程序时优雅地关闭线程池和选择器，并输出相应的信息。

总结

通过以上两个实例，我们可以了解如何使用 Socket 和线程池来实现基本的客户端和服务端通信，以及如何使用多路复用的方式来实现客户端和服务端通信。在实际开发中，我们可以根据具体的需求来进行选择。