一、客户端代码示例
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
类:
-
构造函数:创建
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
。 - 此代码只是一个基础示例,实际应用中可能需要根据具体需求进行扩展,如处理更复杂的消息格式、实现重连机制等。