JAVA异步的TCP 通讯-客户端

一、客户端代码示例

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.AsynchronousSocketChannel;
import java.nio.channels.CompletionHandler;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class AdvancedAsyncTCPClient {
    private static final String SERVER_HOST = "localhost";
    private static final int SERVER_PORT = 8888;
    private static final int BUFFER_SIZE = 1024;
    private final AsynchronousSocketChannel clientChannel;
    private final ExecutorService threadPool;

    public AdvancedAsyncTCPClient() throws IOException {
        // 创建异步套接字通道
        clientChannel = AsynchronousSocketChannel.open();
        // 创建一个固定大小的线程池，用于处理业务逻辑
        threadPool = Executors.newFixedThreadPool(5);
    }

    public void connect() {
        // 异步连接到服务器
        clientChannel.connect(new InetSocketAddress(SERVER_HOST, SERVER_PORT), null, new CompletionHandler<Void, Void>() {
            @Override
            public void completed(Void result, Void attachment) {
                System.out.println("Connected to server: " + SERVER_HOST + ":" + SERVER_PORT);
                // 连接成功后开始读取服务器数据
                startReading();
                // 发送初始消息
                sendMessage("Hello, server!");
            }

            @Override
            public void failed(Throwable exc, Void attachment) {
                System.err.println("Failed to connect to server: " + exc.getMessage());
                closeChannel();
            }
        });
    }

    private void startReading() {
        ByteBuffer buffer = ByteBuffer.allocate(BUFFER_SIZE);
        // 异步读取服务器数据
        clientChannel.read(buffer, buffer, new CompletionHandler<Integer, ByteBuffer>() {
            @Override
            public void completed(Integer bytesRead, ByteBuffer buffer) {
                if (bytesRead > 0) {
                    buffer.flip();
                    byte[] data = new byte[buffer.remaining()];
                    buffer.get(data);
                    String message = new String(data);
                    System.out.println("Received message from server: " + message);
                    // 继续读取服务器数据
                    buffer.clear();
                    clientChannel.read(buffer, buffer, this);
                } else if (bytesRead == -1) {
                    // 服务器关闭连接
                    System.out.println("Server closed the connection");
                    closeChannel();
                }
            }

            @Override
            public void failed(Throwable exc, ByteBuffer buffer) {
                System.err.println("Failed to read data from server: " + exc.getMessage());
                closeChannel();
            }
        });
    }

    public void sendMessage(String message) {
        ByteBuffer buffer = ByteBuffer.wrap(message.getBytes());
        // 异步发送消息到服务器
        clientChannel.write(buffer, buffer, new CompletionHandler<Integer, ByteBuffer>() {
            @Override
            public void completed(Integer bytesWritten, ByteBuffer buffer) {
                if (buffer.hasRemaining()) {
                    // 如果还有数据未发送完，继续发送
                    clientChannel.write(buffer, buffer, this);
                } else {
                    System.out.println("Message sent to server: " + message);
                }
            }

            @Override
            public void failed(Throwable exc, ByteBuffer buffer) {
                System.err.println("Failed to send message to server: " + exc.getMessage());
                closeChannel();
            }
        });
    }

    private void closeChannel() {
        try {
            System.out.println("Closing client connection");
            clientChannel.close();
            threadPool.shutdown();
        } catch (IOException e) {
            System.err.println("Error closing client channel: " + e.getMessage());
        }
    }

    public static void main(String[] args) {
        try {
            AdvancedAsyncTCPClient client = new AdvancedAsyncTCPClient();
            client.connect();
        } catch (IOException e) {
            System.err.println("Error creating client: " + e.getMessage());
        }
    }
}