java selector

java selector使用select輪詢注冊到selector中的channel，如果有channel准備好注冊的事件，select()返回，返回值為可以操作的channel的個數。通過selector.selectedKeys()返回選中的key的集合。遍歷集合中所有的key，判斷key的事件，進行相應的處理，並從集合中remove掉。

客戶端selector的使用邏輯與服務端selector使用邏輯幾乎一致。

Server代碼中監聽了兩個端口8888和8889兩個端口，並注冊到selector中。讀取客戶端發送的信息，並發送信息到客戶端

Client代碼向服務器對應的端口號發送信息，並接收服務器端返回的信息

這里我服務端用了channel+selector客戶端直接用的socket，可以發送給服務端信息，但接收服務端信息沒有接收到（代碼沒有貼出來）

 

PS:使用selector+channel可以在一個線程內監聽多個channel，下面代碼中就監聽了兩個端口的channel

 

SelectorServer.java

import java.io.IOException;
import java.net.InetSocketAddress;
import java.net.ServerSocket;
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;

/**
 * Created by 58 on 2016/11/28.
 */
public class SelectorServer {
    public static void main(String args[]){
        startServer();
    }
    public static ByteBuffer sendBuffer = ByteBuffer.allocate(1024);
    public static ByteBuffer receiveBuffer = ByteBuffer.allocate(1024);

    public static void startServer(){
        //用兩個channel監聽兩個端口
        int listenPort = 8888;
        int listenPort1 = 8889;
        sendBuffer.put("message from server".getBytes());

        try {
            //創建serverchannel，綁定對應的端口
            ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();
            ServerSocket serverSocket = serverSocketChannel.socket();
            InetSocketAddress inetSocketAddress = new InetSocketAddress(listenPort);
            serverSocket.bind(inetSocketAddress);

            //創建第二個channel
            ServerSocketChannel serverSocketChannel1 = ServerSocketChannel.open();
            ServerSocket serverSocket1 = serverSocketChannel1.socket();
            InetSocketAddress inetSocketAddress1 = new InetSocketAddress(listenPort1);
            serverSocket1.bind(inetSocketAddress1);


            //創建selector對象
            Selector selector = Selector.open();

            //設置channel注冊到selector中
            serverSocketChannel.configureBlocking(false);
            serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);

            //第二個channel
            serverSocketChannel1.configureBlocking(false);
            serverSocketChannel1.register(selector, SelectionKey.OP_ACCEPT);

            System.out.println("start to listen port: " + listenPort);
            System.out.println("start to listen port: " + listenPort1);

            //監聽端口
            while(true){
                int readyChannels = selector.select();
                if(readyChannels == 0)
                    continue;
                Set<SelectionKey> selectedKeys = selector.selectedKeys();
                Iterator<SelectionKey> iterator = selectedKeys.iterator();
                while(iterator.hasNext()){
                    SelectionKey selectionKey = iterator.next();
                    dealSelectionKey(selector, selectionKey);

                    iterator.remove();
                }//while
            }//while


        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    public static void dealSelectionKey(Selector selector, SelectionKey selectionKey){
        try{
            //准備好接收新的連接
            if(selectionKey.isAcceptable()){
                ServerSocketChannel serverSocketChannel = (ServerSocketChannel) selectionKey.channel();
                SocketChannel clientSocketChannel = serverSocketChannel.accept();
                clientSocketChannel.configureBlocking(false);
                clientSocketChannel.register(selector, SelectionKey.OP_READ | SelectionKey.OP_WRITE);
                System.out.println("channel is ready acceptable");
            }
            else if(selectionKey.isConnectable()){
                selectionKey.channel().register(selector, SelectionKey.OP_READ);
                System.out.println("channel is connectable.");
            }
            else if(selectionKey.isReadable()){
                //讀去客戶端內容
                SocketChannel clientSocketChannel = (SocketChannel) selectionKey.channel();
                receiveBuffer.clear();
                clientSocketChannel.read(receiveBuffer);
                selectionKey.interestOps(SelectionKey.OP_WRITE);
                System.out.println("message from client is: " + new String(receiveBuffer.array()));
                System.out.println("Thread id : " + Thread.currentThread().getId());
            }
            else if(selectionKey.isWritable()){
                //向客戶端寫數據
                SocketChannel clientSocketChannel = (SocketChannel) selectionKey.channel();
                sendBuffer.flip();
                System.out.println("sendBuffer = " + new String(sendBuffer.array()));
                clientSocketChannel.write(sendBuffer);
                selectionKey.interestOps(SelectionKey.OP_READ);
                System.out.println("channle is writable.");
            }//else if
        }catch (Exception e){

        }
    }
}

SelectorClient.java

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;

/**
 * Created by 58 on 2016/11/28.
 */
public class SelectorClient {
    public static void main(String args[]){
        work();
    }

    public static void work(){
        int serverPort = 8888;
        ByteBuffer sendBuffer = ByteBuffer.wrap("client message".getBytes());
        ByteBuffer receiveBuffer = ByteBuffer.allocate(1024);
        try {
            //創建通道，設置通道注冊到selector中
            SocketChannel socketChannel = SocketChannel.open();
            Selector selector = Selector.open();

            socketChannel.configureBlocking(false);
            socketChannel.register(selector, SelectionKey.OP_WRITE | SelectionKey.OP_READ | SelectionKey.OP_CONNECT);
            socketChannel.connect(new InetSocketAddress("localhost", serverPort));
            int executeTimes = 2;
            //while
            while(executeTimes > 0){
                executeTimes--;
                int reayChannelNum = selector.select();
                if(reayChannelNum == 0){
                    continue;
                }
                Set<SelectionKey> setOfSelectionKey = selector.selectedKeys();
                Iterator<SelectionKey> iterator = setOfSelectionKey.iterator();
                while(iterator.hasNext()){
                    SelectionKey selectionKey = iterator.next();
                    SocketChannel socketChannel1 = (SocketChannel) selectionKey.channel();
                    iterator.remove();
                    if(selectionKey.isConnectable()){
                        if(socketChannel1.isConnectionPending()){
                            socketChannel1.finishConnect();
                            System.out.println("connection complete.");
                            socketChannel1.write(sendBuffer);
                        }
                    }//if isConnectable
                    else if(selectionKey.isReadable()){
                        receiveBuffer.clear();
                        socketChannel1.read(receiveBuffer);
                        receiveBuffer.flip();
                        System.out.println("message from server: " + new String(receiveBuffer.array()));

                    }//else if readable
                    else if(selectionKey.isWritable()){
                        sendBuffer.flip();
                        socketChannel1.write(sendBuffer);
                    }
                }
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
}