一、客户端代码示例
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>() {@Overridepublic void completed(Void result, Void attachment) {System.out.println("Connected to server: " + SERVER_HOST + ":" + SERVER_PORT);// 连接成功后开始读取服务器数据startReading();// 发送初始消息sendMessage("Hello, server!");}@Overridepublic 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>() {@Overridepublic 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();}}@Overridepublic 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>() {@Overridepublic void completed(Integer bytesWritten, ByteBuffer buffer) {if (buffer.hasRemaining()) {// 如果还有数据未发送完，继续发送clientChannel.write(buffer, buffer, this);} else {System.out.println("Message sent to server: " + message);}}@Overridepublic 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 类：
-
构造函数：创建
AsynchronousSocketChannel并初始化一个固定大小的线程池 -
connect()方法：异步连接到服务器，连接成功后开始读取服务器数据并发送初始消息。 -
startReading()方法：异步读取服务器发送的数据，使用CompletionHandler处理读取结果。 -
sendMessage()方法：异步发送消息到服务器，处理可能的未发送完的数据。 -
closeChannel()方法：关闭客户端通道并关闭线程池，处理关闭过程中可能出现的异常。
CompletionHandler：
- 在
connect()、startReading()和sendMessage()方法中使用CompletionHandler来处理异步操作的完成结果。 completed()方法处理操作成功的情况，failed()方法处理操作失败的情况。
线程池：
1.使用 Executors.newFixedThreadPool(5) 创建一个固定大小的线程池，用于处理业务逻辑，避免阻塞 I/O 操作。
三、优点
- 异步 I/O：利用 Java NIO 2 的异步 I/O 特性，提高了客户端的并发处理能力。
- 线程池：使用线程池处理业务逻辑，减少了线程创建和销毁的开销。
- 异常处理：对各种异常情况进行了处理，增强了代码的健壮性。
- 资源管理：在关闭客户端时，正确关闭客户端通道和线程池，避免资源泄漏。
四、注意事项
- 该示例假设服务器运行在
localhost的8888端口，你可以根据实际情况修改SERVER_HOST和SERVER_PORT。 - 此代码只是一个基础示例，实际应用中可能需要根据具体需求进行扩展，如处理更复杂的消息格式、实现重连机制等。
