服务端
- 创建AsynchronousServerSocketChannel，AsynchronousServerSocketChannel.open() 可以自定义线程组
// 创建固定线程池，线程数量
ExecutorService threadPool = Executors.newFixedThreadPool(20);
AsynchronousChannelGroup group = AsynchronousChannelGroup.withThreadPool(threadPool);
// 创建serverSocketChannel ，并用自定义线程组，来处理
AsynchronousServerSocketChannel serverSocketChannel = AsynchronousServerSocketChannel.open(group);
serverSocketChannel.bind(new InetSocketAddress(9998));
- serverSocketChannel.accept(), 为server追加接收监听
serverSocketChannel.accept(null, new AIOServerSocketHandler(serverSocketChannel));
- ServerSocketHandler处理器，用来接收每个客户端的连接
import lombok.extern.slf4j.Slf4j;
import java.nio.ByteBuffer;
import java.nio.channels.AsynchronousServerSocketChannel;
import java.nio.channels.AsynchronousSocketChannel;
import java.nio.channels.CompletionHandler;
/**
* 处理服务端accept的completed完成事件, 即前面定义的serverSocketChannel.accept(null, new AIOServerSocketHandler(serverSocketChannel));
*/
@Slf4j
public class AIOServerSocketHandler implements CompletionHandler<AsynchronousSocketChannel, Void> {
private AsynchronousServerSocketChannel serverSocketChannel;
public AIOServerSocketHandler(AsynchronousServerSocketChannel serverSocketChannel) {
this.serverSocketChannel = serverSocketChannel;
}
// 处理服务端accept的完成连接
@Override
public void completed(AsynchronousSocketChannel socketChannel, Void attachment) {
log.info("server端收到accept请求，并将AIO client channel加入到 AIO Server socket channel中");
serverSocketChannel.accept(attachment, this);
ByteBuffer buffer = ByteBuffer.allocate(1024);
// 这里使用StringBuffer, 将客户端多次请求的数据收集到一起进行处理， 设置read方法的回调处理 Handler
socketChannel.read(buffer, new StringBuffer(), new AIOServerReadHandler(socketChannel, buffer));
}
@Override
public void failed(Throwable exc, Void attachment) {
log.info("failed(Throwable exc, ByteBuffer attachment)");
}
}
- 定义处理客户端连接成功后数据的读取工作
import lombok.extern.slf4j.Slf4j;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.channels.AsynchronousSocketChannel;
import java.nio.channels.CompletionHandler;
/**
* 处理客户端read事件, 即前面定义的socketChannel.read(buffer, new StringBuffer(), new AIOServerReadHandler(socketChannel, buffer));
*/
@Slf4j
public class AIOServerReadHandler implements CompletionHandler<Integer, StringBuffer> {
private AsynchronousSocketChannel socketChannel;
private ByteBuffer buffer;
public AIOServerReadHandler(AsynchronousSocketChannel socketChannel, ByteBuffer buffer) {
this.socketChannel = socketChannel;
this.buffer = buffer;
}
// read事件连接成功
@Override
public void completed(Integer result, StringBuffer attachment) {
if (result == -1) {
try {
this.socketChannel.close();
} catch (IOException e) {
log.info("AIOServerReadHandler socket channel 关闭失败");
throw new RuntimeException(e);
}
}
this.buffer.flip();
byte[] contexts = new byte[1024];
this.buffer.get(contexts, 0, result);
this.buffer.clear();
try {
String msg = new String(contexts, 0, result, "UTF-8");
attachment.append(msg);
log.info("=========输出结果：{}", attachment);
} catch (IOException e) {
log.info("=========输出结果失败", e);
}
if (attachment.indexOf("quit") == -1) {
this.socketChannel.read(this.buffer, attachment, this);
return;
}
log.info("=======收到完整信息，开始处理业务=========");
System.out.println("客户端收到的信息：" + attachment.toString());
attachment = new StringBuffer();
//还要继续监听(一次监听一次通知)
this.socketChannel.read(this.buffer, attachment, this);
}
@Override
public void failed(Throwable exc, StringBuffer attachment) {
try {
socketChannel.close();
} catch (IOException e) {
log.warn("failed, {}", e.fillInStackTrace());
throw new RuntimeException(e);
}
}
}
客户端
import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.AsynchronousSocketChannel;
import java.util.Scanner;
public class Client {
public static void main(String[] args) throws IOException, InterruptedException {
AsynchronousSocketChannel channel = AsynchronousSocketChannel.open();
channel.connect(new InetSocketAddress("127.0.0.1", 9998));
Scanner scanner = new Scanner(System.in);
while (scanner.hasNextLine()) {
String msg = scanner.nextLine();
channel.write(ByteBuffer.wrap(msg.getBytes()));
}
}
}
