public class SimpleServer {
    public static void main(String[] args) throws IOException {
?
        System.out.println("這里是服務(wù)端");
?
        //創(chuàng)建服務(wù)端Channel
        ServerSocketChannel serverChannel = ServerSocketChannel.open();
?
        //設(shè)置非阻塞
        serverChannel.configureBlocking(false);
?
        //創(chuàng)建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();
        //主線程負責(zé)連接
        while (true){
            //當(dāng)沒有事件到來的時候，這里是阻塞的,有事件的時候會自動運行
            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輪詢到的是什么事件，并根據(jù)事件作出回應(yīng)
                //如果是連接事件
                if(key.isAcceptable()){
                    //之前把服務(wù)端channel注冊到selector上時候，把serverChannel放進來了
                    ServerSocketChannel channel = (ServerSocketChannel) key.channel();
                    //接收客戶端channel
                    SocketChannel clientChannel = channel.accept();
                    clientChannel.configureBlocking(false);
                    SelectionKey clientSocketKey = clientChannel.register(selector, 0, clientChannel);
                    //將客戶端channel設(shè)置為可讀事件
                    clientSocketKey.interestOps(SelectionKey.OP_READ);
                    System.out.println("客戶端連接成功"+System.currentTimeMillis());
                    //worker線程開始從客戶端讀數(shù)據(jù)，把客戶端的channel交給worker
                    executor.register(clientChannel,executor);
                    //用channel寫回一條信息
                    clientChannel.write(ByteBuffer.wrap("服務(wù)端寫回客戶端成功".getBytes()));
                    System.out.println("向客戶端發(fā)送數(shù)據(jù)成功"+System.currentTimeMillis());
                }
            }
        }
?
    }
}

public class NioEventLoop extends SingleThreadEventLoop{
?
    protected Selector selector ;
?
    public NioEventLoop() throws IOException {
        this.selector = Selector.open();
    }
?
    public Selector getSelector(){
        return this.selector;
    }
?
    //循環(huán)阻塞，如果有事件發(fā)生，或者隊列有東西，就放行
    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 {
                //等待任務(wù)
                select();
                //I/O任務(wù)來了先處理I/O任務(wù)
                processSelectedKeys();
            } catch (IOException e) {
                throw new RuntimeException(e);
            }finally {
                //最后處理隊列任務(wù)。第一遍走過來的時候會先處理隊列任務(wù)
                runAllTasks();
            }
        }
    }
}

public abstract class SingleThreadEventExecutor implements Executor {
?
 ? ?private volatile boolean isSingle = false;
?
 ? ?private RejectedExecutionHandler rejectedExecutionHandler;
?
 ? ?protected Queue<Runnable> tasksQueue;
?
 ? ?protected Thread thread;
?
 ? ?//初始化的時候，1.構(gòu)建線程池的屬性，如隊列，拒絕策略。2.構(gòu)建當(dāng)前線程的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) {
?
 ? ? ? ?//此時仍然是主線程。先把任務(wù)放入隊列，后續(xù)處理
 ? ? ? ?addTask(task);
 ? ? ? ?System.out.println("任務(wù)添加完成");
 ? ? ? ?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("新線程任務(wù)成功創(chuàng)建");
 ? ? ? ?//這是個異步線程,單線程執(zhí)行器的核心
 ? ? ? ?new Thread(()->{
 ? ? ? ? ? ?thread = Thread.currentThread();
 ? ? ? ? ? ?SingleThreadEventExecutor.this.run();
 ? ? ? ? ? ?System.out.println("新線程任務(wù)跑完了");
 ? ? ?  }).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到當(dāng)前的selector上
    public void register(SocketChannel socketChannel,NioEventLoop nioEventLoop){
        //先判斷當(dāng)前線程是否是執(zhí)行器線程，如果不是，說明是主線程，那么就還沒被構(gòu)造。
        if(inEventLoop(Thread.currentThread())){
            register0(socketChannel,nioEventLoop);
        }else {
            //否則是第一次注冊，就先構(gòu)造線程即可。
            execute(()->{
                register0(socketChannel,nioEventLoop);
                System.out.println("executor執(zhí)行完成");
            });
        }
    }
}