NIO源码解析-SocketChannel

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
public class BIOClientSocket {

    private String address;
    private int port;

    public BIOClientSocket(String address, int port) {
        this.address = address;
        this.port = port;
    }

    public void connectToServer() {
        Socket socket = new Socket();
        try {
            socket.connect(new InetSocketAddress(address, port));

            // 写数据
            new ClientWriteThread(socket).start();
            // 读数据
            BufferedReader bufferedReader = new BufferedReader(new InputStreamReader(socket.getInputStream()));
            String msg = "";
            while ((msg = bufferedReader.readLine()) != null) {
                System.out.println("receive msg: " + msg);
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    public static void main(String[] args) {
        String address = "localhost";
        int port = 9999;

        BIOClientSocket bioClientSocket = new BIOClientSocket(address, port);
        bioClientSocket.connectToServer();
    }
}

/**
 * 客户端发送请求线程
 */
class ClientWriteThread extends Thread {
    private Socket socket;
    private PrintWriter writer;
    private Scanner scanner;

    public ClientWriteThread(Socket socket) throws IOException {
        this.socket = socket;
        this.scanner = new Scanner(System.in);
        this.writer = new PrintWriter(new OutputStreamWriter(socket.getOutputStream()), true);
    }

    @Override
    public void run() {
        String msg = "";

        try {
            // 通过获取对话框里的消息，不断发送到server端
            while ((msg = scanner.nextLine()) != null) {
                if (msg.equals("bye")) {
                    break;
                }
                writer.println(msg);
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
public class NIOClientSocket {

    private String address;
    private int port;
    private Selector selector;

    private ByteBuffer readBuffer = ByteBuffer.allocate(1024);
    private ByteBuffer writeBuffer = ByteBuffer.allocate(1024);

    private Scanner scanner = new Scanner(System.in);


    public NIOClientSocket(String address, int port) throws IOException {
        this.address = address;
        this.port = port;
        this.selector = Selector.open();
    }

    public void connectToServer() {
        try {
            SocketChannel socketChannel = SocketChannel.open();
            socketChannel.configureBlocking(false);

            socketChannel.register(selector, SelectionKey.OP_CONNECT);
            socketChannel.connect(new InetSocketAddress(address, port));

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

    private void connect() {
        while (true) {
            try {
                selector.select();
                Set<SelectionKey> selectionKeys = selector.selectedKeys();
                for (SelectionKey key : selectionKeys) {
                    if (key.isConnectable()) {
                        SocketChannel clientChannel = (SocketChannel) key.channel();
                        if (clientChannel.isConnectionPending()) {
                            clientChannel.finishConnect();
                            System.out.println("client connect success...");
                        }

                        clientChannel.register(selector, SelectionKey.OP_WRITE);
                    } else if (key.isReadable()) {
                        SocketChannel clientChannel = (SocketChannel) key.channel();
                        StringBuffer sb = new StringBuffer("receive msg: ");

                        readBuffer.clear();
                        while (clientChannel.read(readBuffer) > 0) {
                            readBuffer.flip();
                            sb.append(new String(readBuffer.array(), 0, readBuffer.limit()));
                        }

                        System.out.println(sb.toString());
                        clientChannel.register(selector, SelectionKey.OP_WRITE);
                    } else if (key.isWritable()) {
                        SocketChannel clientChannel = (SocketChannel) key.channel();
                        String msg = scanner.nextLine();
                        writeBuffer.clear();

                        writeBuffer.put(msg.getBytes());
                        writeBuffer.flip();
                        clientChannel.write(writeBuffer);
                        clientChannel.register(selector, SelectionKey.OP_READ);
                    }
                }

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

    public static void main(String[] args) {
        String address = "localhost";
        int port = 9999;

        try {
            new NIOClientSocket(address, port).connectToServer();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
}

1
2
3
4
5
// 使当前channel绑定到具体地址
NetworkChannel bind(SocketAddress local) throws IOException;

// 设置socket属性
<T> NetworkChannel setOption(SocketOption<T> name, T value) throws IOException;

1
2
3
4
5
6
// 将当前通道注册到Selector上
public abstract SelectionKey register(Selector sel, int ops, Object att)
        throws ClosedChannelException;

// 获取当前selector上可执行的操作（OP_READ OP_WRITE...）
public final SelectionKey keyFor(Selector sel)

1
2
3
4
5
6
7
8
9
10
// A socket is an endpoint for communication between two machines
public class Socket implements java.io.Closeable {
    /** A socket will have a channel if, and only if, the channel itself was
     * created via the{@link java.nio.channels.SocketChannel#open
     * SocketChannel.open} or {@link
     * java.nio.channels.ServerSocketChannel#accept ServerSocketChannel.accept} */
    public SocketChannel getChannel() {
        return null;
    }
}

1
2
3
4
5
6
7
8
// A selectable channel for stream-oriented connecting sockets
public abstract class SocketChannel
    extends AbstractSelectableChannel
    implements ByteChannel, ScatteringByteChannel, GatheringByteChannel, NetworkChannel
{
    // Retrieves a socket associated with this channel.
	public abstract Socket socket();
}