IO

服务端ServerSocket 客户端Socket

缺点每次客户端建立连接都会另外启一个线程处理。读取和发送数据都是阻塞式的。

如果1000个客户端建立连接将会产生1000个线程

Server端

package bhz.bio.test;

import java.io.IOException;

import java.net.ServerSocket;

import java.net.Socket;

public class Server {

    private int port;

    private ServerSocket serverSocket;

    public Server(int port) {

        this.port = port;

    }

    public void start() throws IOException {

        try {

            serverSocket = new ServerSocket(port);

            while (true) {

               Socket socket = serverSocket.accept();// 阻塞等待客户端建立连接

               new Thread(new ServerHandler(socket)).start();//另外起一個線程處理客戶端的請求

            }

        } catch (IOException e) {

             e.printStackTrace();

            // TODO Auto-generated catch block

            System.out.println("服务器启动失败");

        }finally {

            if(serverSocket!=null) {

                serverSocket.close();//释放资源操作

            }

        }

    }

}

ServerHandler

package bhz.bio.test;

import java.io.BufferedReader;

import java.io.IOException;

import java.io.InputStreamReader;

import java.io.PrintWriter;

import java.net.Socket;

public class ServerHandler implements Runnable {

    private Socket socket;

    public ServerHandler(Socket socket) {

        this.socket = socket;

    }

    @Override

    public void run() {

        // TODO Auto-generated method stub

        BufferedReader in = null;

        PrintWriter out = null;

        try {

            while (true) {

                // 监听客户端的发送消息

                out = new PrintWriter(socket.getOutputStream(),true);

                in = new BufferedReader(new InputStreamReader(socket.getInputStream()));

                String content = in.readLine();

                System.out.println("接收到客户端发送的消息:" + content);//

                out.println("哈哈哈");

            }

        } catch (IOException e) {

            // TODO Auto-generated catch block

            e.printStackTrace();

        } finally {

            if(in != null){

                try {

                    in.close();

                } catch (IOException e) {

                    e.printStackTrace();

                }

            }

            if(out != null){

                try {

                    out.close();

                } catch (Exception e) {

                    e.printStackTrace();

                }

            }

            if(socket != null){

                try {

                    socket.close();

                } catch (IOException e) {

                    e.printStackTrace();

                }

            }

            socket = null;

        }

    }

}

客户端

package bhz.bio.test;

import java.io.BufferedReader;

import java.io.BufferedWriter;

import java.io.IOException;

import java.io.InputStreamReader;

import java.io.PrintWriter;

import java.net.Socket;

import java.net.UnknownHostException;

public class Client {

    private String ip;

    private int port;

    private Socket socket;

    private PrintWriter out;

    private BufferedReader in;

    public Client(String ip, int port) {

        this.ip = ip;

        this.port = port;

    }

    public void connect() {

        try {

            socket = new Socket(ip, port);

            out = new PrintWriter(socket.getOutputStream(),true);

            in=new BufferedReader(new InputStreamReader(socket.getInputStream()));

        } catch (Exception e) {

            // TODO: handle exception

            System.out.println("建立連接失敗");

        }

    }

    public void send(String message) throws IOException {

         out.println(message);

         try {

             String content=in.readLine();

               System.out.println("接收到服务端的数据:"+content);

        } catch (Exception e) {

            // TODO: handle exception

            e.printStackTrace();

        }

    }

}

测试类

package bhz.bio.test;

import java.io.IOException;

public class mainTest {

  public static void main(String[] args) {

      //启动服务端

    new Thread(new Runnable() {

        @Override

        public void run() {

            // TODO Auto-generated method stub

            Server server=new Server(8089);

            try {

                server.start();

            } catch (IOException e) {

                // TODO Auto-generated catch block

                e.printStackTrace();

            }

            try {

                Thread.sleep(Integer.MAX_VALUE);

            } catch (InterruptedException e) {

                // TODO Auto-generated catch block

                e.printStackTrace();

            }

        }

    }).start();

    new Thread(new Runnable() {

        @Override

        public void run() {

            // TODO Auto-generated method stub

            Client client=new Client("127.0.0.1", 8089);

            client.connect();//与服务器建立连接

            try {

                client.send("你好呀");

                client.send("你好呀2");

            } catch (IOException e1) {

                // TODO Auto-generated catch block

                e1.printStackTrace();

            }

            try {

                Thread.sleep(Integer.MAX_VALUE);

            } catch (InterruptedException e) {

                // TODO Auto-generated catch block

                e.printStackTrace();

            }

        }

    }).start();

}

}

输出

接收到客户端发送的消息:你好呀

接收到服务端的数据:哈哈哈

接收到客户端发送的消息:你好呀2

接收到服务端的数据:哈哈哈

使用线程池限制客户端数量(伪异步)

public class Server {

    private int port;

    private ServerSocket serverSocket;

    private ExecutorService executorService;

    public Server(int port) {

        this.port = port;

        executorService=Executors.newFixedThreadPool(50);//限制同时在线客户端数量为10个

    }

    public void start() throws IOException {

        try {

            serverSocket = new ServerSocket(port);

            while (true) {

               Socket socket = serverSocket.accept();// 阻塞等待客户端建立连接

               executorService.execute(new ServerHandler(socket));//线程池处理

            }

        } catch (IOException e) {

             e.printStackTrace();

            // TODO Auto-generated catch block

            System.out.println("服务器启动失败");

        }finally {

            if(serverSocket!=null) {

                serverSocket.close();//释放资源操作

            }

        }

    }

}

Nio

nio改善了io一个客户端建立连接就创建一个线程监听请求的模型。

nio的核心概念

1.Channel( 负责读写数据可以channel读取数据到buffer也可以通过channel写入数据到channel)

FileChannel
DatagramChannel 能通过UDP读写网络中的数据
SocketChannel 能通过TCP读写网络中的数据。
ServerSocketChannel 可以监听新进来的TCP连接，像Web服务器那样。对每一个新进来的连接都会创建一个SocketChannel。

2.Buffers

ByteBuffer
CharBuffer
DoubleBuffer
FloatBuffer
IntBuffer
LongBuffer
ShortBuffer

存储数据的容器

3.Selectors

负责轮询Channel的状态进行对应的操作。发送消息和建立连接都需要往selectors注册channel (一个selectors可以监听无数个channel状态)

buffer的使用

建立一个指定大小的缓冲区

private ByteBuffer readBuf = ByteBuffer.allocate(1024);

往buffer容器添加数据内容position字段都会累加。

读取数据是根据position字段来读取

所以读取之前要记得重置position位置为0

readBuf.flip();

读取缓冲区里面数据

//根据缓冲区的数据长度创建相应大小的byte数组，接收缓冲区的数据

            byte[] bytes = new byte[this.readBuf.remaining()];

            / 接收缓冲区数据

            this.readBuf.get(bytes);

            //打印结果

            String body = new String(bytes).trim();

服務端

package bhz.nio.test;

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;

public class Server {

    private int port;

    private ServerSocketChannel serverSocketChannel;

    private Selector selector;

    //2 建立缓冲区

    private ByteBuffer readBuf = ByteBuffer.allocate(1024);

    private ByteBuffer writeBuf = ByteBuffer.allocate(1024);

    public Server(int port) {

        this.port = port;

    }

    public void start() throws IOException {

        selector=Selector.open();//创建选择器

        serverSocketChannel=ServerSocketChannel.open();//创建监听通道

        serverSocketChannel.configureBlocking(false);//开启非阻塞模式

        serverSocketChannel.socket().bind(new InetSocketAddress(port));//绑定端口

        //把服务器通道注册到多路复用器上，并且监听阻塞事件

        serverSocketChannel.register(this.selector, SelectionKey.OP_ACCEPT);

        //轮询多路复用器

        new Thread(new Runnable() {

            @Override

            public void run() {

                // TODO Auto-generated method stub

                while (true) {

                      //阻塞,只有当至少一个注册的事件发生的时候才会继续.

                    try {

                        selector.select();

                    } catch (IOException e) {

                        // TODO Auto-generated catch block

                        e.printStackTrace();

                    }

                    // 返回多路复用器已经选择的结果集

                    Iterator<SelectionKey> selectionKeys=selector.selectedKeys().iterator();

                    //3 进行遍历

                    while(selectionKeys.hasNext()){

                        //4 获取一个选择的元素

                        SelectionKey key = selectionKeys.next();

                        //5 直接从容器中移除就可以了

                        selectionKeys.remove();

                        //6 如果是有效的

                        if(key.isValid()){

                            //7 如果为阻塞状态

                            if(key.isAcceptable()){

                                accept(key);

                            }

                            //8 如果为可读状态

                            if(key.isReadable()){

                                read(key);

                            }

                            //9 写数据

                            if(key.isWritable()){

                                //this.write(key); //ssc

                            }

                        }

                    }

                }

            }

        }).start();

    }

    private void accept(SelectionKey key) {

        try {

            //1 获取服务通道

            ServerSocketChannel ssc =  (ServerSocketChannel) key.channel();

            //2 执行阻塞方法

            SocketChannel sc = ssc.accept();

            //3 设置阻塞模式

            sc.configureBlocking(false);

            //4 注册到多路复用器上，并设置读取标识

            sc.register(this.selector, SelectionKey.OP_READ);

        } catch (IOException e) {

            e.printStackTrace();

        }

    }

    private void read(SelectionKey key) {

        try {

            //1 清空缓冲区旧的数据

            this.readBuf.clear();

            //2 获取之前注册的socket通道对象

            SocketChannel sc = (SocketChannel) key.channel();

            //3 读取数据

            int count = sc.read(this.readBuf);

            //4 如果没有数据

            if(count == -1){

                key.channel().close();

                key.cancel();

                return;

            }

            //5 有数据则进行读取 读取之前需要进行复位方法(把position 和limit进行复位)

            this.readBuf.flip();

            //6 根据缓冲区的数据长度创建相应大小的byte数组，接收缓冲区的数据

            byte[] bytes = new byte[this.readBuf.remaining()];

            //7 接收缓冲区数据

            this.readBuf.get(bytes);

            //8 打印结果

            String body = new String(bytes).trim();

            System.out.println("Server : " + body);

            // 9..可以写回给客户端数据

        } catch (IOException e) {

            e.printStackTrace();

        }

    }

}

客戶端

package bhz.nio.test;

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;

public class Client {

    private SocketChannel socketChannel;

    private ByteBuffer readBuf = ByteBuffer.allocate(1024);

    private ByteBuffer writeBuf = ByteBuffer.allocate(1024);

    private Selector selector;

    public void connect(String ip,int port) throws IOException {

        selector=Selector.open();//创建选择器

        socketChannel=SocketChannel.open();//创建连接

         if(socketChannel.connect(new InetSocketAddress(ip,port)));

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

         socketChannel.configureBlocking(false);

        socketChannel.register(selector, SelectionKey.OP_CONNECT); //注册到多路复用器

        new Thread(new Runnable() {

            @Override

            public void run() {

                // TODO Auto-generated method stub

                   SelectionKey key = null;

                while (true) {

                    try {

                        selector.select();

                    } catch (IOException e) {

                        // TODO Auto-generated catch block

                        e.printStackTrace();

                    }

                    Iterator<SelectionKey> selectionKeys=selector.selectedKeys().iterator();

                    while (selectionKeys.hasNext()) {

                        key=selectionKeys.next();

                        if(key.isReadable()) {

                        }

                    }

                }

            }

        }).start();

    }

    private void read(SelectionKey key) {

        try {

            //1 清空缓冲区旧的数据

            this.readBuf.clear();

            //2 获取之前注册的socket通道对象

            SocketChannel sc = (SocketChannel) key.channel();

            //3 读取数据

            int count = sc.read(this.readBuf);

            //4 如果没有数据

            if(count == -1){

                key.channel().close();

                key.cancel();

                return;

            }

            //5 有数据则进行读取 读取之前需要进行复位方法(把position 和limit进行复位)

            this.readBuf.flip();

            //6 根据缓冲区的数据长度创建相应大小的byte数组，接收缓冲区的数据

            byte[] bytes = new byte[this.readBuf.remaining()];

            //7 接收缓冲区数据

            this.readBuf.get(bytes);

            //8 打印结果

            String body = new String(bytes).trim();

            System.out.println("client : " + body);

            // 9..可以写回给客户端数据

        } catch (IOException e) {

            e.printStackTrace();

        }

    }

    public void write(String message) throws IOException {

        byte[] bytes =message.getBytes();

        //把数据放到缓冲区中

        writeBuf.put(bytes);

        //对缓冲区进行复位

        writeBuf.flip();

        //写出数据

        socketChannel.write(writeBuf);

        //清空缓冲区数据

        writeBuf.clear();

    }

}

測試類

package bhz.nio.test;

import java.io.IOException;

public class TestMain {

    public static void main(String[] args) throws InterruptedException {

        final Server server = new Server(8088);

        final Client client=new Client();

        new Thread(new Runnable() {

            @Override

            public void run() {

                // TODO Auto-generated method stub

                try {

                    server.start();

                } catch (IOException e) {

                    // TODO Auto-generated catch block

                    e.printStackTrace();

                }

                try {

                    Thread.sleep(Integer.MAX_VALUE);

                } catch (InterruptedException e) {

                    // TODO Auto-generated catch block

                    e.printStackTrace();

                }

            }

        }).start();

        Thread.sleep(2000);

        new Thread(new Runnable() {

            @Override

            public void run() {

                // TODO Auto-generated method stub

                try {

                    client.connect("127.0.0.1", 8088);

                    client.write("哈哈哈哈");

                } catch (IOException e) {

                    // TODO Auto-generated catch block

                    e.printStackTrace();

                }

            }

        }).start();

    }

}

AIO

服務端類AsynchronousServerSocketChannel

客戶端類AsynchronousSocketChannel

服务端

public class Server {

    private int port;

    //服务器通道

     AsynchronousServerSocketChannel serverSocketChannel;

    //线程组

    private AsynchronousChannelGroup threadGroup;

    private ExecutorService executorService;

    public Server(int port) {

        this.port=port;

    }

    public void start() throws IOException {

        //创建线程池

        executorService=Executors.newCachedThreadPool();

        //创建线程组

        threadGroup = AsynchronousChannelGroup

                    .withThreadPool(executorService);

        //创建服务器通道

        serverSocketChannel = AsynchronousServerSocketChannel.open(threadGroup);

        //进行绑定

        serverSocketChannel.bind(new InetSocketAddress(port));

        serverSocketChannel.accept(this, new ServerHandler());

    }

}

异步读handle

/**

 * 客户端接收服务器数据回调

 * @author Administrator

 *

 */

public class ClientReadHandler implements CompletionHandler<Integer, ByteBuffer> {

    @Override

    public void completed(Integer result, ByteBuffer attachment) {

        // TODO Auto-generated method stub

        attachment.flip();//读模式 position设置为0

        byte[] datas=new byte[attachment.remaining()];

        attachment.get(datas);

        try {

            System.out.println("服务端响应"+new String(datas,"utf-8"));

        } catch (UnsupportedEncodingException e) {

            // TODO Auto-generated catch block

            e.printStackTrace();

        }

    }

    @Override

    public void failed(Throwable exc, ByteBuffer attachment) {

        // TODO Auto-generated method stub

        System.out.println("客户端读取失败");

    }

}

客户端

package bnz.aio.test;

import java.io.IOException;

import java.io.Writer;

import java.net.InetSocketAddress;

import java.nio.ByteBuffer;

import java.nio.channels.AsynchronousSocketChannel;

import java.nio.channels.CompletionHandler;

public class Client implements CompletionHandler<Void, Client> {

    private String ip;

    private int port;

    private AsynchronousSocketChannel asc;

    public Client(String ip, int port) {

        this.ip = ip;

        this.port = port;

    }

    public void connect() throws IOException {

        asc = AsynchronousSocketChannel.open();

        asc.connect(new InetSocketAddress(ip, port), this, this);

          //读取数据

        ByteBuffer readBuffer = ByteBuffer.allocate(1024);

         asc.read(readBuffer,readBuffer,new ClientReadHandler());//服务器写入数据回调

    }

    public void Writer(String message) {

        byte[] req = message.getBytes();

        ByteBuffer writeBuffer = ByteBuffer.allocate(req.length);

        writeBuffer.put(req);

        writeBuffer.flip();

        //异步写

        asc.write(writeBuffer,writeBuffer, new WriteHandler(asc));

    }

    //连接成功的回调

    @Override

    public void completed(Void result, Client attachment) {

        // TODO Auto-generated method stub

        System.out.println("建立连接成功");

    }

    //异常回调

    @Override

    public void failed(Throwable exc, Client attachment) {

        // TODO Auto-generated method stub

        System.out.println("建立连接失败");

    }

}

读handle

/**

 * 客户端接收服务器数据回调

 * @author Administrator

 *

 */

public class ClientReadHandler implements CompletionHandler<Integer, ByteBuffer> {

    @Override

    public void completed(Integer result, ByteBuffer attachment) {

        // TODO Auto-generated method stub

        attachment.flip();//读模式 position设置为0

        byte[] datas=new byte[attachment.remaining()];

        attachment.get(datas);

        try {

            System.out.println("服务端响应"+new String(datas,"utf-8"));

        } catch (UnsupportedEncodingException e) {

            // TODO Auto-generated catch block

            e.printStackTrace();

        }

    }

    @Override

    public void failed(Throwable exc, ByteBuffer attachment) {

        // TODO Auto-generated method stub

        System.out.println("客户端读取失败");

    }

}

写handle

/**

 * 客户端异步写回调

 * @author Administrator

 *

 */

public class WriteHandler implements CompletionHandler<Integer, ByteBuffer> {

    private AsynchronousSocketChannel channel;

     public WriteHandler(AsynchronousSocketChannel asynchronousSocketChannel) {

        // TODO Auto-generated constructor stub

         this.channel=asynchronousSocketChannel;

    }

    @Override

    public void completed(Integer result, ByteBuffer attachment) {

        // TODO Auto-generated method stub

        attachment.hasRemaining();

        channel.write(attachment,attachment,this);

    }

    @Override

    public void failed(Throwable exc, ByteBuffer attachment) {

        // TODO Auto-generated method stub

        System.out.println("数据写入失败");

    }

}

io nio aio比较

io 同步阻塞 jdk4以前的网络通信模型，io一个连接需要开启一个线程处理，在大量连接情况下服务器会创建N个线程。导致cpu线程调度频繁性能底下 (同步阻塞)

nio 同步非阻塞 jdk4之后的网络通信,主要是为了解决io模型下一个线程处理一个连接的性能问题，引入事件驱动。l客户端连接服务端注册到复用器，然后轮询复用器注册连接的状态。只需要一个线程处理 (同步非阻塞)

aio 异步非阻塞 jdk7推出的网络通信框架跟nio不同的是读写都是异步的(完全交给内核处理) 完成后通知应用程序

同步阻塞可以理解为买票排队。没买到票之前什么也不能做。

同步非阻塞可以理解为邮局。经常需要跑去邮局去询问有没有自己的邮件。

异步非阻塞可以理解为去饭店点餐然后回家饭店做好打包后送上门

应用场景

io可以应用在少量且固定连接的网络通信应为io的模型更简单容易理解

nio 在大量连接情况使用

aio在大量连接且读写数据非常大的情况下

jdk4

java IO(BIO)、NIO、AIO

IO