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

二、代码分析

AdvancedAsyncTCPClient 类：

1. 构造函数：创建 AsynchronousSocketChannel 并初始化一个固定大小的线程池

2. connect() 方法：异步连接到服务器，连接成功后开始读取服务器数据并发送初始消息。

3. startReading() 方法：异步读取服务器发送的数据，使用 CompletionHandler 处理读取结果。

4. sendMessage() 方法：异步发送消息到服务器，处理可能的未发送完的数据。

5. closeChannel() 方法：关闭客户端通道并关闭线程池，处理关闭过程中可能出现的异常。

CompletionHandler：

1. 在 connect()、startReading() 和 sendMessage() 方法中使用 CompletionHandler 来处理异步操作的完成结果。
2. completed() 方法处理操作成功的情况，failed() 方法处理操作失败的情况。

线程池：

        1.使用 Executors.newFixedThreadPool(5) 创建一个固定大小的线程池，用于处理业务逻辑，避免阻塞 I/O 操作。

三、优点

• 异步 I/O：利用 Java NIO 2 的异步 I/O 特性，提高了客户端的并发处理能力。
• 线程池：使用线程池处理业务逻辑，减少了线程创建和销毁的开销。
• 异常处理：对各种异常情况进行了处理，增强了代码的健壮性。
• 资源管理：在关闭客户端时，正确关闭客户端通道和线程池，避免资源泄漏。

四、注意事项

• 该示例假设服务器运行在 localhost 的 8888 端口，你可以根据实际情况修改 SERVER_HOST 和 SERVER_PORT。
• 此代码只是一个基础示例，实际应用中可能需要根据具体需求进行扩展，如处理更复杂的消息格式、实现重连机制等。