Netty-JAVA基礎實現,NIO基礎

時間  2020-05-15
標籤 netty java 基礎 實現 nio 欄目 Netty 简体版
原文   原文鏈接

BIO 是JAVA網絡通訊中同步阻塞的實現方式，NIO是JAVA的同步非阻塞方式，大體示意以下
Netty-JAVA基礎實現,NIO基礎java

每一個客戶端以socketchannel（能夠視同bio下的socket）向服務器發送鏈接或者請求，服務器端在啓動時建立一個ServerSocketChannel，用於綁定服務的端口和IP以及處理鏈接到socket的請求.同時ServerSocketChannel也註冊在selector下。windows

當selector以指定的時間間隔進行輪詢時，若是發現有新的請求進來，會根據狀況採起兩個方向的處理。一個是新建鏈接的請求，由ServerSocketChannel經過accept去生成一個新的socketChannel，並註冊到selector上。或者在輪詢時發現個某個socketchannel知足了某類處理要求(如讀數據)，則進行相應的處理處理數組

下附相關代碼
客戶端
package netty.nio;緩存

public class ClientThreadMain {服務器

public static void main(String[] args) {

    String ip = "127.0.0.1";
    int port = 9999;

    new Thread(new NIOSocketClient(ip,port)).start();

}

}網絡

package netty.nio;less

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

/**ide

  • NIO模式客戶端實現
  • @author zhousjmas@hotmail.com

  • */
    public class NIOSocketClient implements Runnable{this

    private String ip;
    private int port;
    private boolean running = true;
    private Selector selector;
    private SocketChannel socketChannel;

    //初始化時完成ip,port selector與socketChannel的設置
    public NIOSocketClient(String ip,int port) {
    this.ip = ip;
    this.port = port;

    try {
    selector = Selector.open();
    socketChannel = SocketChannel.open();
    //設置爲非阻塞的
    socketChannel.configureBlocking(false);
} catch (IOException e) {
    // TODO Auto-generated catch block
    e.printStackTrace();
}

    }

    @Override
    public void run() {
    //發起鏈接請求
    connect();
    while(running) {

    /*
     * 超時毫秒數
     * timeout --If positive, block for up to timeoutmilliseconds,
     *  more or less, while waiting for achannel to become ready; 
     * if zero, block indefinitely;must not be negative
     */
    try {
        selector.select(1000);
        //迭代處理selector中全部的SelectionKey
        Set<SelectionKey> selectionKeys = selector.selectedKeys();

        Iterator<SelectionKey> it = selectionKeys.iterator();
        SelectionKey key = null;
        while(it.hasNext()) {

            key = it.next();
            it.remove();
            handler(key);
        }

    } catch (IOException e) {

        e.printStackTrace();
    }
}

    }

    //處理器，處理當鏈接成功與收到返回成功後的各自處理業務
    private void handler(SelectionKey selectionKey) {

    if(selectionKey.isValid()) {

    SocketChannel sc = (SocketChannel)selectionKey.channel();
    //若是已經聯通,那麼開始發送數據
    if(selectionKey.isConnectable()) {
        if(sc.isConnected()) {
            try {
                sc.register(selector, SelectionKey.OP_READ);
                sendMessage(sc);
                selectionKey.cancel();
                try {
                    sc.close();
                } catch (IOException e) {
                    // TODO Auto-generated catch block
                    e.printStackTrace();
                }
            } catch (ClosedChannelException e) {
                // TODO Auto-generated catch block
                e.printStackTrace();
            }
        }
        //若是能夠讀取,那麼開始進行讀取
        if(selectionKey.isReadable()) {
            ByteBuffer buffer = ByteBuffer.allocate(1024);
            try {
                int bufferSize = sc.read(buffer);
                if(bufferSize>0) {
                    //按照這個長度建立一個比特數組接收傳入的數據
                    //The number of elements remaining in this buffer
                    byte[] bytes = new byte[buffer.remaining()];
                    //轉成實際的字符串打印出來
                    String info  = new String(bytes,"UTF-8");
                    System.out.println("傳回的信息爲:"+info);

                }else {
                    selectionKey.cancel();
                    sc.close();
                }
            } catch (IOException e) {
                // TODO Auto-generated catch block
                e.printStackTrace();
            }
        }
    }
}

    }

    /*

    • 發送消息
      */
      private void sendMessage(SocketChannel sc) {

      byte[] bytes = "HELLO JAVA NIO".getBytes();
      ByteBuffer buffer = ByteBuffer.allocateDirect(bytes.length);
      buffer.put(bytes);
      buffer.flip();
      try {
      sc.write(buffer);
      if(!buffer.hasRemaining()) {
      System.out.println("已經發送信息");
      }
      } catch (IOException e) {
      // TODO Auto-generated catch block
      e.printStackTrace();
      }
      }

    /*

    • 一啓動線程就發起服務端請求
      */
      private void connect() {

      try {
      socketChannel.connect(new InetSocketAddress(ip,port));
      if(socketChannel.finishConnect()) {
      //若是連通即將socketChannel註冊到selector上，關注的事件是讀
      socketChannel.register(selector, SelectionKey.OP_READ);
      //向服務端發送數據
      sendMessage(socketChannel);
      }

      else {
    socketChannel.register(selector, SelectionKey.OP_CONNECT);
}

      } catch (IOException e) {
      // TODO Auto-generated catch block
      e.printStackTrace();
      }
      }

}
上邊代碼能夠看出客戶端的Selector也在輪詢確認服務端的返回狀況，並進行處理

如下是服務端實現

package netty.nio;

/**

  • NIO模式服務端實現
  • @author zhousjmas@hotmail.com

  • */
    public class NIOSocketServer {

    public static void main(String[] args) {

    int port = 9999;

NIOSocketSelector selector = new NIOSocketSelector(port);

new Thread(selector).start();

    }

}

package netty.nio;

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.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.util.Iterator;
import java.util.Set;

/**

  • NIO的多路複用器的實現
  • @author zhousjmas@hotmail.com

  • */
    public class NIOSocketSelector implements Runnable{

    //多路複用器
    private Selector selector;
    //設通道綁定隊列的最大長度
    private int backlog = 800;
    //任務持續執行標誌
    private boolean running = true;

    private int requestNum =0;

    //通道
    private ServerSocketChannel serverSocketChannel;

    //傳入端口參數
    public NIOSocketSelector(int port) {

    try {
    //設置多路複用器
    selector = Selector.open();
    //設置通道
    serverSocketChannel = ServerSocketChannel.open();
    /* 設置通道爲非阻 塞模式
     * block If true then this channel will be placed inblocking mode; 
     * if false then it will be placednon-blocking mode
     */
    serverSocketChannel.configureBlocking(false);
    /* 最大隊列長度
     * backlog - requested maximum length of the queue of incoming connections
     */
    serverSocketChannel.socket().bind( new InetSocketAddress(port), backlog);

    serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);

} catch (IOException e) {
    // TODO Auto-generated catch block
    e.printStackTrace();
}

    }

    @Override
    public void run() {

    while(running) {
    //System.out.println("啓動服務端 監聽請求");

    /*
     * 超時毫秒數
     * timeout --If positive, block for up to timeoutmilliseconds,
     *  more or less, while waiting for achannel to become ready; 
     * if zero, block indefinitely;must not be negative
     */
    try {
        //每秒進行一次輪詢處理
        selector.select(1000);
        //迭代處理selector中全部的SelectionKey
        Set<SelectionKey> selectionKeys = selector.selectedKeys();

        Iterator<SelectionKey> it = selectionKeys.iterator();
        SelectionKey key = null;
        while(it.hasNext()) {

            key = it.next();
            it.remove();
            handler(key);
        }

    } catch (IOException e) {

        e.printStackTrace();
    }

}
if(selector!=null) {
    try {
        selector.close();
    }catch(IOException ie) {

    }
}

    }

    private void handler(SelectionKey selectionKey) {

    //若是當前的Key可用，取到當前key對應的channel
if(selectionKey.isValid()) {
    ServerSocketChannel ssc;
    SocketChannel sc;
    //若是是新來的請求就新建一條socketchannel綁定到selector
    if(selectionKey.isAcceptable()) {

        ssc = (ServerSocketChannel)selectionKey.channel();
        try {

            sc = ssc.accept();
            //設置爲非阻塞模式
            sc.configureBlocking(false);
            //註冊到Selector下，設定爲讀打開
            sc.register(selector, SelectionKey.OP_READ);
        } catch (IOException e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        }
    }

    //若是是能夠讀取數據了，那就進行數據讀取
    if(selectionKey.isReadable()) {

        requestNum  = requestNum +1;
        System.out.println("當前請求數:"+requestNum);
        sc = (SocketChannel)selectionKey.channel();
        ByteBuffer buffer = ByteBuffer.allocate(1024);
        try {
            //判斷讀入數據的長度
            int bufferSize = sc.read(buffer);
            if(bufferSize>0) {
                //將緩存字節數組的指針設置爲數組的開始序列即數組下標0
                //這樣從buffer的頭部開始對buffer進行遍歷 
                buffer.flip();
                //按照這個長度建立一個比特數組接收傳入的數據
                //The number of elements remaining in this buffer
                byte[] bytes = new byte[buffer.remaining()];
                buffer.get(bytes);
                //轉成實際的字符串打印出來
                String info  = new String(bytes,"UTF-8");
                System.out.println("傳入的信息爲:"+info);
                //對傳入的信息進行響應返回
                response(sc,new Long(System.currentTimeMillis()).toString());
            }else if(bufferSize<0){
                selectionKey.cancel();
                sc.close();
            }
        } catch (IOException e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        }
    }

}

    }

    /*

    • 處理回寫的數據
      */
      private void response(SocketChannel socketChannel,String msg) {

      if(msg!=null && msg.trim().length()>0) {
      byte[] bytes = msg.getBytes();
      //按字符串長度新建ByteBuffer
      ByteBuffer buffer = ByteBuffer.allocate(bytes.length) ;
      buffer.put(bytes);
      buffer.flip();
      try {
      //將返回的數據信息寫入到SocketChannel中
      socketChannel.write(buffer);
      } catch (IOException e) {
      // TODO Auto-generated catch block
      e.printStackTrace();
      }
      }
      }

    public void isRunning(boolean running) {
    this.running = running;
    }

}

這裏注意幾個地方，一個是backlog，這個值在windows下設置爲大於等於1000後會拋一個異常。須要將之調小，目前懷疑的是當對端 口的監聽大於必定數量，會報這個異常

第二是running雖然寫了，可是並無用。只用於提供一個可能性，能夠在某些必要的時候打斷這種運行

第三是在IDE下進行調試時最好在兩個IDE下進行調試。不要把服務端與客戶端都啓動在一個IDE下，會出現某名的異常

相關文章
相關標籤/搜索
每日一句
    每一个你不满意的现在，都有一个你没有努力的曾经。
本站公眾號
   歡迎關注本站公眾號,獲取更多信息
聯繫我們 最近搜索 最新文章 沪ICP备13005482号-10 MyBatis教程 SQL 教程 MySQL教程 Java 教程 Thymeleaf 教程 Hibernate教程 Spring教程 Redis教程