Java NIO

本來想寫關(guān)于netty類的時序圖，學(xué)習(xí)下設(shè)計模式并學(xué)習(xí)如何擴展Java nio的，畢竟對于我這種擰螺絲釘?shù)慕o我一個任務(wù)如何寫出高內(nèi)聚低耦合的代碼才是重要的，但是找不到合適相關(guān)聯(lián)Java NIO和netty相關(guān)的代碼，所以我花費了一點時間整理了下相關(guān)代碼。

import java.io.IOException;
import java.net.InetSocketAddress;
import java.net.SocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.*;
import java.util.Iterator;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicInteger;

public class NioSocketServer {

    private static final int NCPU = Runtime.getRuntime().availableProcessors();

    private static final int BUFFERSIZE = 1024;

    private final ServerSocketChannel serverSocketChannel;

    //簡易的Reactor模型，一個boss線程，2倍核數(shù)的工作線程
    private final Selector bossselector;
    private final Work[] works;
    private AtomicInteger index = new AtomicInteger();

    //用于緩存每個客戶端粘包拆包等數(shù)據(jù).
    private Map<SocketAddress,Read> cacheChannelBuffer = new ConcurrentHashMap<>();

    public NioSocketServer()throws IOException{
        serverSocketChannel = ServerSocketChannel.open();
        serverSocketChannel.configureBlocking(false);
        serverSocketChannel.socket().bind(new InetSocketAddress(8888));
        bossselector = Selector.open();
        SelectionKey selectionKey = serverSocketChannel.register(bossselector, SelectionKey.OP_ACCEPT);
        new Thread(new Boss()).start();
        works = new Work[NCPU * 2];
        for(int i = 0;i < works.length;i++){
            new Thread(works[i] = new Work(Selector.open(),i)).start();
        }
    }

    public void accept(SelectionKey key) {
        System.out.println("accept事件");
        try {
            ServerSocketChannel serverSocketChannel = (ServerSocketChannel) key.channel();
            SocketChannel socketChannel = serverSocketChannel.accept();
            socketChannel.configureBlocking(false);
            int i = index.getAndIncrement() & works.length - 1;
            works[i].register(socketChannel);
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
    //包協(xié)議：包=包頭(4byte)+包體，包頭內(nèi)容為包體的數(shù)據(jù)長度
    public void read(SelectionKey selectionKey) {
        System.out.println("read事件");
        try {
            SocketChannel channel = (SocketChannel) selectionKey.channel();
            SocketAddress address = channel.getRemoteAddress();
            Read read = cacheChannelBuffer.get(address);
            int bodyLen = -1;
            ByteBuffer byteBuffer;
            if(read == null){
                byteBuffer = ByteBuffer.allocate(BUFFERSIZE);
            }else{
                if(read.headerLength == -1){
                    byteBuffer = ByteBuffer.allocate(BUFFERSIZE);
                }else{
                    bodyLen = read.getHeaderLength();
                    byteBuffer = ByteBuffer.allocate(read.getHeaderLength());
                }
                ByteBuffer readByteBuffer = read.getByteBuffer();
                if(readByteBuffer != null && readByteBuffer.hasRemaining()){
                    readByteBuffer.flip();
                    byteBuffer.put(readByteBuffer);
                }
                read.setByteBuffer(null);
                read.setHeaderLength(-1);
            }
            channel.read(byteBuffer);
            byteBuffer.flip();
            while (byteBuffer.remaining() > 0) {
                if (bodyLen == -1) {// 還沒有讀出包頭，先讀出包頭
                    if (byteBuffer.remaining() >= 4) {// 讀出包頭，否則緩存
                        byteBuffer.mark();
                        bodyLen = byteBuffer.getInt();
                    } else {
                        remaining(read, byteBuffer, address, bodyLen);
                        break;
                    }
                } else {// 已經(jīng)讀出包頭
                    if (byteBuffer.remaining() >= bodyLen) {// 大于等于一個包，否則緩存
                        byte[] bodyByte = new byte[bodyLen];
                        byteBuffer.get(bodyByte, 0, bodyLen);
                        bodyLen = -1;
                        System.out.println("receive from clien content is:" + new String(bodyByte));
                    } else {
                        remaining(read, byteBuffer, address, bodyLen);
                        break;
                    }
                }
            }
//            String httpResponse = "HTTP/1.1 200 OK\r\n" + "Content-Length: 38\r\n" + "Content-Type: text/html\r\n" + "<html><body>Hello World!</body></html>";
//            ByteBuffer buffer = ByteBuffer.wrap(httpResponse.getBytes());
//            int len = channel.write(buffer);
//            if (len < 0){
//                throw new IllegalArgumentException();
//            }
//            if (len == 0) {
//                selectionKey.interestOps(SelectionKey.OP_WRITE);
//            }

            selectionKey.interestOps(SelectionKey.OP_READ);
        } catch (Exception e) {
            try {
                selectionKey.cancel();
                serverSocketChannel.close();
            } catch (IOException e1) {
                e1.printStackTrace();
            }
            e.printStackTrace();
        }
    }
    private void remaining(Read read,ByteBuffer byteBuffer,SocketAddress address,int bodyLen){
        if(!byteBuffer.hasRemaining()){
            return;
        }
        if(read == null){
            read = new Read();
            cacheChannelBuffer.put(address,read);
        }
        read.setHeaderLength(bodyLen);
        int remaining = byteBuffer.remaining();
        byte[] remainingByte = new byte[remaining];
        byteBuffer.get(remainingByte, 0, remaining);
        read.setByteBuffer(ByteBuffer.allocate(remaining).put(remainingByte));
    }
    public void write(SelectionKey selectionKey) {
        System.out.println("write事件");
        SocketChannel socketChannel = (SocketChannel) selectionKey.channel();
        String httpResponse = "HTTP/1.1 200 OK\r\n" + "Content-Length: 38\r\n" + "Content-Type: text/html\r\n" + "<html><body>Hello World!</body></html>";
        System.out.println("response from server to client");
        try {
            ByteBuffer byteBuffer = ByteBuffer.wrap(httpResponse.getBytes());
            while (byteBuffer.hasRemaining()) {
                socketChannel.write(byteBuffer);
            }
            selectionKey.interestOps(SelectionKey.OP_READ);
        } catch (IOException e) {
            try {
                selectionKey.cancel();
                serverSocketChannel.close();
            } catch (IOException e1) {
                e1.printStackTrace();
            }
            e.printStackTrace();
        }
    }
    class Boss implements Runnable{

        private volatile boolean isExit;

        public boolean isExit() {
            return isExit;
        }

        public void setExit(boolean exit) {
            isExit = exit;
        }
        @Override
        public void run(){
            try {
                while (!isExit) {
                    int selectKey = bossselector.select();
                    if (selectKey > 0) {
                        Set<SelectionKey> keySet = bossselector.selectedKeys();
                        Iterator<SelectionKey> iter = keySet.iterator();
                        while (iter.hasNext()) {
                            SelectionKey selectionKey = iter.next();
                            iter.remove();
                            if (selectionKey.isAcceptable()) {
                                accept(selectionKey);
                            } else {
                                System.out.println("boss線程不可能出現(xiàn)work線程的事件，請檢查代碼。");
                            }
                        }
                    }
                }
            } catch (IOException e) {
                e.printStackTrace();
            }finally {

            }
        }
    }

    class Work implements Runnable{

        private final Selector workSelector;
        private final int i;
        private volatile boolean isExit ;

        public Work(Selector workSelector,int i){
            this.workSelector = workSelector;
            this.i = i;
        }
        public SelectionKey register(SocketChannel socketChannel) throws ClosedChannelException {
            return socketChannel.register(workSelector, SelectionKey.OP_READ);
        }

        public boolean isExit() {
            return isExit;
        }

        public void setExit(boolean exit) {
            isExit = exit;
        }

        @Override
        public void run(){
            try {
                while (!isExit) {
                    int selectKey = workSelector.select(10);
                    if (selectKey > 0) {
                        Set<SelectionKey> keySet = workSelector.selectedKeys();
                        Iterator<SelectionKey> iter = keySet.iterator();
                        while (iter.hasNext()) {
                            SelectionKey selectionKey = iter.next();
                            iter.remove();
                            if (selectionKey.isAcceptable()) {
                                System.out.println("work線程不可能出現(xiàn)boss線程的事件，請檢查代碼。");
                            }else if (selectionKey.isReadable()) {
                                read(selectionKey);
                            }else if (selectionKey.isWritable()) {
                                write(selectionKey);
                            }
                        }
                    }
                }
            } catch (IOException e) {
                e.printStackTrace();
            }finally {

            }
        }
    }

    class Read{
        private ByteBuffer byteBuffer;
        private int headerLength = -1;

        public ByteBuffer getByteBuffer() {
            return byteBuffer;
        }

        public void setByteBuffer(ByteBuffer byteBuffer) {
            this.byteBuffer = byteBuffer;
        }

        public int getHeaderLength() {
            return headerLength;
        }

        public void setHeaderLength(int headerLength) {
            this.headerLength = headerLength;
        }
    }

    public static void main(String args[]) throws IOException {
        NioSocketServer server = new NioSocketServer();
    }
}

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

public class NioSocketClient  {
    private SocketChannel socketChannel;
    private Selector selector = null;

    public NioSocketClient() throws IOException{
        InetSocketAddress inetSocketAddress = new InetSocketAddress(8888);
        selector = Selector.open();
        socketChannel = SocketChannel.open();
        socketChannel.configureBlocking(false);
        socketChannel.connect(inetSocketAddress);
        socketChannel.register(selector, SelectionKey.OP_CONNECT);
        new Thread(new Work()).start();
    }

    public void finishConnect(SelectionKey key) {
        System.out.println("client finish connect!");
        SocketChannel socketChannel = (SocketChannel) key.channel();
        try {
            socketChannel.finishConnect();
            key.interestOps(SelectionKey.OP_WRITE);
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
    public void read(SelectionKey key) throws IOException {
        SocketChannel channel = (SocketChannel) key.channel();
        ByteBuffer byteBuffer = ByteBuffer.allocate(1024);
        int len = channel.read(byteBuffer);
        if (len > 0) {
            byteBuffer.flip();
            byte[] byteArray = new byte[byteBuffer.limit()];
            byteBuffer.get(byteArray);
            System.out.println("client receive from server,content:"+new String(byteArray));
            len = channel.read(byteBuffer);
            byteBuffer.clear();
        }
        key.interestOps(SelectionKey.OP_READ);
    }

    public void send(SelectionKey key) {
        SocketChannel channel = (SocketChannel) key.channel();
        for (int i = 0; i < 10; i++) {
            String ss = i + "Server ,how are you ? this is package message from NioSocketClient!";
            int headSize  = (ss).getBytes().length;
            ByteBuffer byteBuffer = ByteBuffer.allocate(4 + headSize);
            byteBuffer.putInt(headSize);
            byteBuffer.put(ss.getBytes());
            byteBuffer.flip();
            System.out.println("client send:" + i + ",context:"  + ss);
            while (byteBuffer.hasRemaining()) {
                try {
                    channel.write(byteBuffer);
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
        }
        key.interestOps(SelectionKey.OP_READ);
    }

    class Work implements Runnable{
        @Override
        public void run(){
            while (true) {
                try {
                    int key = selector.select();
                    if (key > 0) {
                        Set<SelectionKey> keySet = selector.selectedKeys();
                        Iterator<SelectionKey> iter = keySet.iterator();
                        while (iter.hasNext()) {
                            SelectionKey selectionKey = null;
                            synchronized (iter) {
                                selectionKey = iter.next();
                                iter.remove();
                            }

                            if (selectionKey.isConnectable()) {
                                finishConnect(selectionKey);
                            }
                            if (selectionKey.isWritable()) {
                                send(selectionKey);
                            }
                            if (selectionKey.isReadable()) {
                                read(selectionKey);
                            }
                        }
                    }
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
        }
    }

    public static void main(String args[]) throws IOException {
        NioSocketClient client = new NioSocketClient();
    }
}

以上服務(wù)端實現(xiàn)了簡易的Reactor模型，并自定義了通信協(xié)議（協(xié)議格式：長度+內(nèi)容），并實現(xiàn)了粘包和拆包的邏輯。netty不管怎么封裝，最終都是要封裝NIO那幾行代碼。
以下是我遇到的問題和我覺得需要注意細(xì)節(jié)。
1.我是在win10 jdk1.8實現(xiàn)的代碼，服務(wù)端worker線程獲取數(shù)據(jù)采用select()時，當(dāng)有新的客戶端連接時，獲取不到數(shù)據(jù)，采用select(long timeout)可以獲取到數(shù)據(jù)，這個問題沒找到原因。
2.Java底層無法得知channel獲取了多少數(shù)據(jù)，所以需要自定義ByteBuffer的大小，在發(fā)生拆包粘包時需要注意。netty實現(xiàn)了自動實現(xiàn)計算ByteBuffer的大小，不一定準(zhǔn)確。
3.自定義的nio代碼中，很少看到OP_WRITE的處理，經(jīng)?？吹降拇a就是在請求處理完成后，直接通過下面的代碼將結(jié)果返回給客戶端。什么時候采用OP_WRITE，引用別人的一段話：
如果客戶端的網(wǎng)絡(luò)或者是中間交換機的問題，使得網(wǎng)絡(luò)傳輸?shù)男屎艿?，這時候會出現(xiàn)服務(wù)器已經(jīng)準(zhǔn)備好的返回結(jié)果無法通過TCP/IP層傳輸?shù)娇蛻舳?。這時候在執(zhí)行上面這段程序的時候就會出現(xiàn)以下情況。
(1) bb.hasRemaining()一直為“true”，因為服務(wù)器的返回結(jié)果已經(jīng)準(zhǔn)備好了。
(2) socketChannel.write(bb)的結(jié)果一直為0，因為由于網(wǎng)絡(luò)原因數(shù)據(jù)一直傳不過去。
(3) 因為是異步非阻塞的方式，socketChannel.write(bb)不會被阻塞，立刻被返回。
(4) 在一段時間內(nèi)，這段代碼會被無休止地快速執(zhí)行著，消耗著大量的CPU的資源。事實上什么具體的任務(wù)也沒有做，一直到網(wǎng)絡(luò)允許當(dāng)前的數(shù)據(jù)傳送出去為止。
因此，要對OP_WRITE加以處理，常用用法為：

            String httpResponse = "HTTP/1.1 200 OK\r\n" + "Content-Length: 38\r\n" + "Content-Type: text/html\r\n" + "<html><body>Hello World!</body></html>";
            ByteBuffer buffer = ByteBuffer.wrap(httpResponse.getBytes());
            int len = channel.write(buffer);
            if (len < 0){
                throw new IllegalArgumentException();
            }
            if (len == 0) {
                selectionKey.interestOps(SelectionKey.OP_WRITE);
            }

以上這段話在我實現(xiàn)的服務(wù)端中屏蔽了，后續(xù)會講解netty時序圖，學(xué)習(xí)netty優(yōu)秀的源碼。
最后，目前在找工作，現(xiàn)在在家cha

?著作權(quán)歸作者所有,轉(zhuǎn)載或內(nèi)容合作請聯(lián)系作者
【社區(qū)內(nèi)容提示】社區(qū)部分內(nèi)容疑似由AI輔助生成，瀏覽時請結(jié)合常識與多方信息審慎甄別。
平臺聲明：文章內(nèi)容（如有圖片或視頻亦包括在內(nèi)）由作者上傳并發(fā)布，文章內(nèi)容僅代表作者本人觀點，簡書系信息發(fā)布平臺，僅提供信息存儲服務(wù)。

相關(guān)閱讀更多精彩內(nèi)容

友情鏈接更多精彩內(nèi)容