NIO编程

1、最基本的NIO编程

通过不停的轮询，来查看是否有请求。

public class Server1 {public static void main(String[] args) throws IOException {ByteBuffer byteBuffer = ByteBuffer.allocate(16);//ServerSocketChannel ssc = ServerSocketChannel.open();ssc.configureBlocking(false);ssc.bind(new InetSocketAddress(8080));List<SocketChannel> channels = new ArrayList<>();while (true) {SocketChannel sc = ssc.accept();if (sc != null) {channels.add(sc);}for (SocketChannel channel : channels) {channel.read(byteBuffer);byteBuffer.flip();System.out.println(Charset.defaultCharset().decode(byteBuffer).toString());byteBuffer.clear();}}}
}

2、通过Selector实现NIO编程

public class Server2 {public static void main(String[] args) throws IOException {     // 0. 创建、 配置链接ServerSocketChannel ssc = ServerSocketChannel.open();ssc.configureBlocking(false);ssc.bind(new InetSocketAddress(8080));// 1. 创建SelectorSelector selector = Selector.open();// 2. 建立 selector 和 channel 之间的联系 （注册）SelectionKey sscKey = ssc.register(selector, 0, null);// 3. 设置 SelectionKey 关注事件sscKey.interestOps(SelectionKey.OP_ACCEPT);while (true) {// 4. 调用selector方法// selector 方法： 如果有事件，就继续执行后续代码， 如果没有事件，就阻塞；有未处理事件，不会阻塞selector.select();// 5. 处理事件Iterator<SelectionKey> iter = selector.selectedKeys().iterator();while (iter.hasNext()) {SelectionKey key = iter.next();// 手动删除 已经处理完事件的key ！！！iter.remove();System.out.println("key: " + key);// 区分事件类型if (key.isAcceptable()) {ServerSocketChannel channel = (ServerSocketChannel) key.channel();// key.cancel(); // 不想处理，取消事件// 6. 注册 SocketChannelSocketChannel sc = channel.accept();sc.configureBlocking(false);SelectionKey scKey = sc.register(selector, 0, null);scKey.interestOps(SelectionKey.OP_READ);System.out.println("建立链接：" + sc);}else if (key.isReadable()) {try {SocketChannel channel = (SocketChannel) key.channel();ByteBuffer byteBuffer = ByteBuffer.allocate(16);/*如果正常退出，read = -1*/int read = channel.read(byteBuffer);if (read == -1) {key.cancel();}else {// 未处理消息byteBuffer.flip();System.out.println(Charset.defaultCharset().decode(byteBuffer).toString());}} catch (IOException e) {e.printStackTrace();key.cancel();}}else {System.out.println("others");}}}
}

3、使用多线程，优化Selector方案

public class Server5 {public static void main(String[] args) throws IOException {// 设置当前线程名Thread.currentThread().setName("boss");// 设置 Selector bossSelector boss = Selector.open();// 设置 ServerSocketChannelServerSocketChannel ssc = ServerSocketChannel.open();ssc.configureBlocking(false);ssc.bind(new InetSocketAddress(1234));SelectionKey bossKey = ssc.register(boss, 0, null);bossKey.interestOps(SelectionKey.OP_ACCEPT);// 1. 创建 Worker// Worker worker0 = new Worker("worker-0");// Worker[] workers = new Worker[2];// 获取CPU的核心数Worker[] workers = new Worker[Runtime.getRuntime().availableProcessors()];// 在docker中 容器不是物理隔离的，会拿到cpu的个数，而不是容器申请时的个数。for (int i = 0; i < workers.length; i++) {workers[i] = new Worker("worker-" + i);}AtomicInteger index = new AtomicInteger();while (true){boss.select();Iterator<SelectionKey> iterator = boss.selectedKeys().iterator();while (iterator.hasNext()) {SelectionKey key = iterator.next();iterator.remove();if (key.isAcceptable()) {ServerSocketChannel channel = (ServerSocketChannel) key.channel();SocketChannel sc = channel.accept();sc.configureBlocking(false);System.out.println("connected..." +  sc.getRemoteAddress());// 2. 关联 worker//round robin 负载均衡workers[index.getAndIncrement() % workers.length].register(sc);// worker0.register(sc);}}}}static class Worker implements Runnable{private Thread thread;private Selector selector;private String name;private volatile boolean start = false; // 还未初始化public ConcurrentLinkedQueue<Runnable> queue = new ConcurrentLinkedQueue<>();public Worker(String name) {this.name = name;}public void register(SocketChannel sc) throws IOException {if (!start) {selector = Selector.open();thread = new Thread(this, name);thread.start();start = true;}queue.add(()->{try {sc.register(selector, SelectionKey.OP_READ, null);} catch (ClosedChannelException e) {e.printStackTrace();}});selector.wakeup(); // 唤醒 select() 方法// 方法二： 更简单的方法, 在这个进行注册// sc.register(selector, SelectionKey.OP_READ, null);}@Overridepublic void run() {while (true) {try {selector.select();Runnable task = queue.poll();if (task != null) {task.run(); // 在这个位置执行注册}Iterator<SelectionKey> iterator = selector.selectedKeys().iterator();while (iterator.hasNext()) {SelectionKey key = iterator.next();iterator.remove();if (key.isReadable()) {ByteBuffer buffer = ByteBuffer.allocate(16);SocketChannel channel = (SocketChannel) key.channel();channel.read(buffer);buffer.flip();System.out.println(Charset.defaultCharset().decode(buffer));}}} catch (IOException e) {e.printStackTrace();}}}}
}