Java NIO案例

Posted 2020-10-25 .x->y=z

服务端、客户端

package com.dsp.nio;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.util.Iterator;
import java.util.Set;

/**
 *
 * 监控是否可连接、可读、可写
 *
 * 代码中巧妙使用了SocketChannel的attach功能，将Hanlder和可能会发生事件的channel链接在一起，当发生事件时，可以立即触发相应链接的Handler
 *
 */
public class Reactor implements Runnable {

    private static Logger log = LoggerFactory.getLogger(Reactor.class);

    final Selector selector;

    final ServerSocketChannel serverSocket;

    /**
     * 服务端配置初始化，监听端口
     * @param port
     * @throws IOException
     */
    public Reactor(int port) throws IOException {
        this.selector = Selector.open();
        this.serverSocket = ServerSocketChannel.open();
        this.serverSocket.socket().bind(new InetSocketAddress(port));
        this.serverSocket.configureBlocking(false);
        SelectionKey selectionKey = this.serverSocket.register(selector, SelectionKey.OP_ACCEPT);
        // 利用selectionKey的attache功能绑定Acceptor 如果有事情，触发Acceptor
        selectionKey.attach(new Acceptor());
        log.info("===>>> attach(new Acceptor())");
    }
    /*
     * SPI
     */
    // Alternatively, use explicit SPI provider
    // SelectorProvider selectorProvider = SelectorProvider.provider();
    // selector = selectorProvider.openSelector();
    // serverSocket = selectorProvider.openServerSocketChannel();

    /**
     * 分发请求
     *
     * @param selectionKey
     */
    void dispatch(SelectionKey selectionKey) {
        Runnable run = (Runnable) (selectionKey.attachment());
        if (run != null) {
            run.run();
        }
    }

    /**
     * 监听连接和channel是否就绪
     */
    public void run() {
        try {
            /**
             * 线程未被中断
             */
            while (!Thread.interrupted()) {
                int readySize = this.selector.select();
                log.info("I/O ready size = {}", readySize);
                Set<?> selectedKeys = this.selector.selectedKeys();
                Iterator<?> iterator = selectedKeys.iterator();
                // Selector如果发现channel有OP_ACCEPT或READ事件发生，下列遍历就会进行。
                while (iterator.hasNext()) {
                    /*
                     * 一个新的连接，第一次出发Accepter线程任务，之后触发Handler线程任务
                     */
                    SelectionKey selectionKey = (SelectionKey) iterator.next();
                    log.info("===>>> acceptable = {}, connectable = {}, readable = {}, writable = {}.", 
　　　　　　　　　　　　　　selectionKey.isAcceptable(), selectionKey.isConnectable(), 
　　　　　　　　　　　　　　selectionKey.isReadable(), selectionKey.isWritable());
                    dispatch(selectionKey);
                }
                selectedKeys.clear();
            }
        } catch (IOException ex) {
            log.info("reactor stop!" + ex);
        }
    }

    /**
     * 处理新连接
     *
     * @author dsp
     *
     */
    class Acceptor implements Runnable {

        @Override
        public void run() {
            try {
                log.debug("===>>> ready for accept!");
                SocketChannel socketChannel = serverSocket.accept();
                if (socketChannel != null) {
                    new Handler(selector, socketChannel);
                }
            } catch (IOException ex) {
                /* . . . */
            }
        }

    }

}

package com.dsp.nio;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.SocketChannel;

/**
 *
 * 处理读写
 *
 */
final class Handler implements Runnable {

    private static Logger log = LoggerFactory.getLogger(Reactor.class);

    static final int MAX_IN = 1024;

    static final int MAX_OUT = 1024;

    ByteBuffer inputBuffer = ByteBuffer.allocate(MAX_IN);

    ByteBuffer output = ByteBuffer.allocate(MAX_OUT);

    final SocketChannel socketChannel;

    final SelectionKey selectionKey;

    static final int READING = 0, SENDING = 1;

    int state = READING;

    /**
     * 注意在Handler里面又执行了一次attach，覆盖前面的Acceptor，下次该Handler又有READ事件发生时，将直接触发Handler，从而开始了数据的
     * “读 -> 处理 -> 写 -> 发出” 等流程处理。
     *
     * @param selector
     * @param socketChannel
     * @throws IOException
     */
    Handler(Selector selector, SocketChannel socketChannel) throws IOException {
        this.socketChannel = socketChannel;
        this.socketChannel.configureBlocking(false);
        this.selectionKey = this.socketChannel.register(selector, 0);
        this.selectionKey.attach(this);
        this.selectionKey.interestOps(SelectionKey.OP_READ);
        // selector.wakeup();
    }

    /**
     * 只是返回true，具体的判断没有实现
     *
     * @return
     */
    boolean inputIsComplete() {
        return true;
    }

    /**
     * 只是返回true，具体的判断没有实现
     *
     * @return
     */
    boolean outputIsComplete() {
        return true;
    }

    /**
     * 处理数据（无具体实现）
     */
    void process(String msg) {
//        output.put("hello world, hello dsp！".getBytes());
        String outMsg = "out + " + msg;
        output.put(outMsg.getBytes());
        output.flip();
    }

    /**
     * 读取请求数据并处理
     *
     * @throws IOException
     */
    void read() throws IOException {
        log.info("===>>> read into bytebuffer from socketchannel inputs.");
        if (inputIsComplete()) {
            socketChannel.read(inputBuffer);
            inputBuffer.flip();
            byte[] inputBytes = new byte[inputBuffer.limit()];
            inputBuffer.get(inputBytes);
            String inputString = new String(inputBytes);
            log.info("===>>> 从客户端读取请求信息 = {}", inputString);
            log.info("===>>> read complete.");

            process(inputString);

            state = SENDING;
            // 读完了数据之后，注册OP_WRITE事件
            selectionKey.interestOps(SelectionKey.OP_WRITE);
        }
    }

    /**
     * 返回响应信息
     *
     * @throws IOException
     */
    void send() throws IOException {
        log.info("===>>> write into socketchannel from bytebuffer outputs");
        socketChannel.write(output);
        if (outputIsComplete()) {
            // The key will be removed from all of the selector‘s key sets during the next
            // selection operation.
            selectionKey.cancel();
            // 关闭通过，也就关闭了连接
            socketChannel.close();
            log.info("===>>> close socketchannel after write complete");
        }
    }

    @Override
    public void run() {
        try {
            if (state == READING)
                read();
            else if (state == SENDING)
                send();
        } catch (IOException ex) {
            /* . . . */
        }
    }

}

package com.dsp.nio;

import java.io.IOException;

/**
 *
 * Model: Reactor in SingleThread
 *
 * 利用NIO多路复用机制，多路IO复用一个线程
 *
 * @author dsp
 *
 */
public class ReactorInSingleThreadServer {

    public static void main(String args[]) throws IOException {
        Reactor reactor = new Reactor(9999);
        reactor.run(); // 不会开启线程，相当于普通方法调用
    }

}

package com.dsp.nio;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.SocketChannel;
import java.util.concurrent.LinkedBlockingQueue;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 *
 * 访问NIO服务器的客户端
 *
 * @author dsp
 *
 */
public class ReactorInSingleThreadClient extends Thread {

    private static Logger log = LoggerFactory.getLogger(ReactorInSingleThreadClient.class);

    private static LinkedBlockingQueue<Thread> failureQueue = new LinkedBlockingQueue<Thread>();

    @Override
    public void run() {
        try {
            ByteBuffer buffer = ByteBuffer.allocate(1024);
            SocketChannel socketChannel = SocketChannel.open();
            boolean connected = socketChannel.connect(new InetSocketAddress(9999));
            if (connected) {
                log.info("===>>> 和服务器 {} 已连接...", socketChannel.getRemoteAddress());
                /*
                 * 请求
                 */
                String msg = "in  + 你好，dsp！" + Thread.currentThread().getName();
                buffer.put(msg.getBytes());
                buffer.flip();
                socketChannel.write(buffer);
                buffer.clear();

                /*
                 * 响应
                 */
                buffer.clear();
                socketChannel.read(buffer);
                buffer.flip();
                byte[] data = new byte[buffer.limit()];
                buffer.get(data);
                String string = new String(data);
                log.info("===>>> " + string);
                buffer.clear();

                socketChannel.close();
            } else {
                log.error("连不上服务器...");
            }
        } catch (java.net.ConnectException e) {
            failureQueue.offer(this);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    public static void main(String[] args) throws IOException, InterruptedException {
        int maxThreads = 3000;
        while (maxThreads-- > 0) {
            new Thread(new ReactorInSingleThreadClient()).start();
        }

        Thread.sleep(Integer.MAX_VALUE);
    }

}

^_^