AIO demo

服务端

1. 创建AsynchronousServerSocketChannel，AsynchronousServerSocketChannel.open() 可以自定义线程组

// 创建固定线程池，线程数量
ExecutorService threadPool = Executors.newFixedThreadPool(20);
AsynchronousChannelGroup group = AsynchronousChannelGroup.withThreadPool(threadPool);
// 创建serverSocketChannel ，并用自定义线程组，来处理
AsynchronousServerSocketChannel serverSocketChannel = AsynchronousServerSocketChannel.open(group);
serverSocketChannel.bind(new InetSocketAddress(9998));

2. serverSocketChannel.accept(), 为server追加接收监听

serverSocketChannel.accept(null, new AIOServerSocketHandler(serverSocketChannel));

3. 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)");
    }
}

4. 定义处理客户端连接成功后数据的读取工作

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