NiO simple example

One 、BIO and NIO

        Java Common in IO The collection has BIO and NIO Two kinds of .

        BIO( Synchronous blocking IO)： Conventional Java IO Programming , Its basic classes and interfaces are java.io In bag , The server implements a pattern of one thread per connection , That is, when the client has a connection request, the server needs to start an independent Acceptor The thread is responsible for listening for client connections , And in while(true) Listen for requests initiated by the client in the loop , If this connection doesn't do anything, it will cause unnecessary thread overhead , It is suitable for a small and fixed number of connections , This service mode requires high price for server resources , And limited to applications , yes JDK1.4 The only choice before , The program is simple and easy to understand . The schematic diagram is as follows ：

        NIO( Synchronous nonblocking IO)：  Related classes are placed in java.nio Under the bag and its subpackages , Its core component is ：Selector( Selectors )、Channel( passageway )、Buffer( buffer ). comparison BIO oriented stream Stream programming ,NIO It's for Buffer Buffer programmed , It's much lower , adopt Stream You can get the response Channel,channel And buffer Combined operation , When buffer When the thread is not full, it does not need to block here , You can do other things , wait for buffer Read or write data when it is full , therefore NIO It's non blocking ; And compared with Stream for , because channel It's two-way , So combine buffer Two way operation is possible , and Stream Only one-way operation ;Selector Used to process multiple channels with a single thread , It circularly listens to multiple client connection channels , If there is no data in the channel, that is, when the client has no request, it can deal with other channels or do other things , If there is data in the channel, he will select this channel and process it , This enables one thread to handle multiple connections . The schematic diagram is as follows ：

Two 、NIO Code example  

          Reference resources netty Medium NIO Realization , Network programming , Generally exist read 、 Write 、 Connection event , The frequency of read and write events is high , Connection events occur only when the client initiates a connection . Refer to this feature ,netty Use different... For connection events and read-write events Selector To listen to , Make the server more efficient , The following is a rough implementation ：

BossThread： Used to listen for client connection events

@Slf4j
public class BossThreadServer {

    public static void main(String[] args) throws IOException {
        Thread.currentThread().setName("bossGroup");
        ServerSocketChannel ssc = ServerSocketChannel.open();
        ssc.configureBlocking(false);
        Selector boss = Selector.open();
        ssc.register(boss, SelectionKey.OP_ACCEPT);
        ssc.bind(new InetSocketAddress(8002));
        // Create a fixed number of worker To handle read and write events , And carry on worker The initialization 
        // The first version is recommended to be set to the server CPU The core number 
        //bug：Runtime.getRuntime().availableProcessors() If it's deployed in docker Under the container , Because the container is not physically isolated , So this method will get the physical machine CPU Number , Instead of container application CPU Number 
        // The problem lies in jdk10 Just fix it , Use jvm Parameters UseContainerSupport To configure , Default on 
        WorkerV2[] workers = new WorkerV2[Runtime.getRuntime().availableProcessors()];
        for (int i = 0; i < workers.length; i++) {
             workers[i] = new WorkerV2("worker-" + i);
        }
        AtomicInteger sum = new AtomicInteger();
        while (true){
            boss.select();
            Iterator<SelectionKey> keyIterator = boss.selectedKeys().iterator();
            while (keyIterator.hasNext()){
                SelectionKey key = keyIterator.next();
                keyIterator.remove();
                if (key.isAcceptable()){
                    SocketChannel socketChannel = ssc.accept();
                    socketChannel.configureBlocking(false);
                    log.info("connected...{}", socketChannel.getRemoteAddress());
                    // relation worker Of selector
                    log.info("before register...{}", socketChannel.getRemoteAddress());
                    // initialization worker And add a registration event 
                    workers[sum.incrementAndGet()%workers.length].register(socketChannel);
                    log.info("after register...{}", socketChannel.getRemoteAddress());
                }
            }
        }
    }
}

Worker： Handling read-write Events

/**
 * 1.0 Version USES selector.select(long times) To solve boss In the thread socketChannel Register and worker Starting selector The priority of blocking , But it will cause resource consumption 
 *  therefore 2.0 Version will be socketChannel Register on worker In the initialization , And use message synchronization queue to realize , When the initialization worker when , Submit a registration to the queue sc The task of , stay worker Thread 
 * run Method to listen to the queue , To perform registration sc The task of 
 * @create 2022/7/14 6:03 PM
 */
@Slf4j
public class WorkerV2 implements Runnable{
    private Thread thread;
    public Selector selector;
    private String name;
    // Whether it is starting 
    private volatile boolean start = false;
    // Synchronize message queues 
    private ConcurrentLinkedQueue<Runnable> queue = new ConcurrentLinkedQueue<>();

    public WorkerV2(String name){
        this.name = name;
    }

    /**
     *  Initialize threads and selector
     */
    public void register(SocketChannel sc) throws IOException {
        if (!start) {
            thread = new Thread(this, name);
            thread.start();
            selector = Selector.open();
            start = true;
        }
        // Add a task to the queue , But this task was not carried out immediately 
        queue.add(() -> {
            try {
                sc.register(selector, SelectionKey.OP_READ);
            } catch (ClosedChannelException e) {
                e.printStackTrace();
            }
        });
        //wakeup and select Be similar to park and unpark, Realized by semaphore , Don't care about the order 
        selector.wakeup();
    }

    /**
     * worker Specifically monitor read and write events 
     */
    @Override
    public void run() {
        while (true){
            try {
                selector.select();
                // perform 
                Runnable task = queue.poll();
                if (task != null){task.run();}
                Iterator<SelectionKey> iterator = selector.selectedKeys().iterator();
                while (iterator.hasNext()){
                    SelectionKey key = iterator.next();
                    iterator.remove();
                    if (key.isReadable()){
                        ByteBuffer buffer = ByteBuffer.allocate(16);
                        SocketChannel channel = (SocketChannel) key.channel();
                        channel.read(buffer);
                        log.info("read data...{}", channel.getRemoteAddress());
                        buffer.flip();
                    }
                }
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }
}

TestClient: Test client

/**
 * @create 2022/7/19 2:15 PM
 */
public class TestClient {

    public static void main(String[] args) throws IOException {
        SocketChannel socketChannel = SocketChannel.open();
        socketChannel.connect(new InetSocketAddress("localhost", 8002));
        socketChannel.write(Charset.defaultCharset().encode("hello world"));
        System.in.read();
    }
}