java Socket编程实现I/O多路复用的示例

public class SimpleServer {
    public static void main(String[] args) throws IOException {
​
        System.out.println("这里是服务端");
​
        //创建服务端Channel
        ServerSocketChannel serverChannel = ServerSocketChannel.open();
​
        //设置非阻塞
        serverChannel.configureBlocking(false);
​
        //创建Selector
        Selector selector = Selector.open();
​
        //0代表不对任何事件感兴趣
        SelectionKey selectionKey = serverChannel.register(selector, 0, serverChannel);
​
        //对连接接收事件感兴趣
        selectionKey.interestOps(SelectionKey.OP_ACCEPT);
​
        //绑定端口
        serverChannel.bind(new InetSocketAddress(8080));
​
        NioEventLoop executor = new NioEventLoop();
        //主线程负责连接
        while (true){
            //当没有事件到来的时候，这里是阻塞的,有事件的时候会自动运行
            selector.select();
            //如果有事件到来，这里可以得到注册到该selector上的所有的key，每一个key上都有一个channel
            Set<SelectionKey> selectionKeys = selector.selectedKeys();
            //得到集合的迭代器
            Iterator<SelectionKey> keyIterator = selectionKeys.iterator();
            while (keyIterator.hasNext()){
                //得到每一个key
                SelectionKey key = keyIterator.next();
                //首先要从集合中把key删除，否则会一直报告该key
                keyIterator.remove();
                //接下来就要处理事件，判断selector轮询到的是什么事件，并根据事件作出回应
                //如果是连接事件
                if(key.isAcceptable()){
                    //之前把服务端channel注册到selector上时候，把serverChannel放进来了
                    ServerSocketChannel channel = (ServerSocketChannel) key.channel();
                    //接收客户端channel
                    SocketChannel clientChannel = channel.accept();
                    clientChannel.configureBlocking(false);
                    SelectionKey clientSocketKey = clientChannel.register(selector, 0, clientChannel);
                    //将客户端channel设置为可读事件
                    clientSocketKey.interestOps(SelectionKey.OP_READ);
                    System.out.println("客户端连接成功"+System.currentTimeMillis());
                    //worker线程开始从客户端读数据，把客户端的channel交给worker
                    executor.register(clientChannel,executor);
                    //用channel写回一条信息
                    clientChannel.write(ByteBuffer.wrap("服务端写回客户端成功".getBytes()));
                    System.out.println("向客户端发送数据成功"+System.currentTimeMillis());
                }
            }
        }
​
    }
}

public class NioEventLoop extends SingleThreadEventLoop{
​
    protected Selector selector ;
​
    public NioEventLoop() throws IOException {
        this.selector = Selector.open();
    }
​
    public Selector getSelector(){
        return this.selector;
    }
​
    //循环阻塞，如果有事件发生，或者队列有东西，就放行
    private void select() throws IOException {
        while (true){
            //阻塞等待事件，如果3s都没有事件过来，可能是没有初始化。
            int select = selector.select(3000);
            if(select != 0 || hasTasks()){
                break;
            }
        }
    }
​
    private void runAllTasks() {
​
        for (;;){
            Runnable task = tasksQueue.poll();
            if(task == null){
                break;
            }
            System.out.println("开始处理注册事件");
            task.run();
        }
    }
​
    private void processSelectedKeys() throws IOException {
        System.out.println("开始处理I/O事件");
        Iterator<SelectionKey> iterator = selector.selectedKeys().iterator();
        while (iterator.hasNext()){
            SelectionKey key = iterator.next();
            iterator.remove();
            //由于连接事件被处理完了，只剩下只读事件了
            if (key.isReadable()) {
                SocketChannel socketChannel = (SocketChannel) key.channel();
                ByteBuffer buffer = ByteBuffer.allocate(1024);
                int len = 0;
                len = socketChannel.read(buffer);
                if (len == -1) {
                    socketChannel.close();
                    break;
                } else {
                    buffer.flip();
                    System.out.println(Charset.defaultCharset().decode(buffer).toString() + System.currentTimeMillis());
                }
            }
        }
    }
​
    @Override
    public void run(){
        while (true) {
            try {
                //等待任务
                select();
                //I/O任务来了先处理I/O任务
                processSelectedKeys();
            } catch (IOException e) {
                throw new RuntimeException(e);
            }finally {
                //最后处理队列任务。第一遍走过来的时候会先处理队列任务
                runAllTasks();
            }
        }
    }
}

public abstract class SingleThreadEventExecutor implements Executor {
​
    private volatile boolean isSingle = false;
​
    private RejectedExecutionHandler rejectedExecutionHandler;
​
    protected Queue<Runnable> tasksQueue;
​
    protected Thread thread;
​
    //初始化的时候，1.构建线程池的属性，如队列，拒绝策略。2.构建当前线程的Selector
    public SingleThreadEventExecutor() {
        this.tasksQueue = new LinkedBlockingQueue<>(Integer.MAX_VALUE);
        this.rejectedExecutionHandler = new ThreadPoolExecutor.AbortPolicy();
    }
​
    protected final void reject(Runnable task) {
//        rejectedExecutionHandler.rejectedExecution(task, this);
    }
​
    protected boolean inEventLoop(Thread thread){
        return this.thread == thread;
    }
​
    protected boolean hasTasks(){
        return this.tasksQueue.isEmpty();
    }
​
​
    @Override
    public void execute(Runnable task) {
​
        //此时仍然是主线程。先把任务放入队列，后续处理
        addTask(task);
        System.out.println("任务添加完成");
        startThread(task);
    }
​
    protected final void addTask(Runnable task){
        this.tasksQueue.add(task);
    }
​
    private void startThread(Runnable task) {
        if (isSingle) {
            return;
        }
        isSingle = true;
        System.out.println("新线程任务成功创建");
        //这是个异步线程,单线程执行器的核心
        new Thread(()->{
            thread = Thread.currentThread();
            SingleThreadEventExecutor.this.run();
            System.out.println("新线程任务跑完了");
        }).start();
    }
​
​
    protected abstract void run();
​
​
}

public abstract class SingleThreadEventLoop extends SingleThreadEventExecutor{
​
    private void register0(SocketChannel socketChannel,NioEventLoop nioEventLoop){
        try {
            socketChannel.configureBlocking(false);
            socketChannel.register(nioEventLoop.getSelector(), SelectionKey.OP_READ);
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }
​
    //注册socketChannel到当前的selector上
    public void register(SocketChannel socketChannel,NioEventLoop nioEventLoop){
        //先判断当前线程是否是执行器线程，如果不是，说明是主线程，那么就还没被构造。
        if(inEventLoop(Thread.currentThread())){
            register0(socketChannel,nioEventLoop);
        }else {
            //否则是第一次注册，就先构造线程即可。
            execute(()->{
                register0(socketChannel,nioEventLoop);
                System.out.println("executor执行完成");
            });
        }
    }
}