曰本孕妇做爰网站,深圳坪山高级中学,免费的黄金软件,医疗器械网站怎么做趁着三天假期#xff0c;把Java NIO和Reactor模式整理总结了下#xff0c;文章特别细节的知识点没有写#xff0c;如一些API的具体实现。类似数据读到Buffer后再写出时#xff0c;为什么需要复位操作#xff0c;这些都属于NIO基础知识#xff0c;是学习Reactor模式的前置…趁着三天假期把Java NIO和Reactor模式整理总结了下文章特别细节的知识点没有写如一些API的具体实现。类似数据读到Buffer后再写出时为什么需要复位操作这些都属于NIO基础知识是学习Reactor模式的前置条件。1. 原始Ractor模式相关组件的解释Handle(句柄或是描述符)本质上表示一种资源是操作系统提供的该资源用于表示一个个事件,比如文件描述符或者是针对于网络编程中的Socket描述符。事件既可以来自于外部也可以来自内部外部事件比如说客户端的连接请求客户端发送过来数据等内部事件比如说操纵系统产生的定时器事件等。它本质上就是一个文件描述符。Handle是事件产生的发源地。Synchronous Event Demultiplexer(同步事件分离器)它本身是一个系统调用用于等待事件的发生(事件可能是一个也可能是多个)。调用方在调用它的时候会被阻塞一直阻塞到同步事件分离器上有事件产生为止。对于Linux来说同步事件分离器指的就是常用的I/O多路复用机制比如说select、poll、epoll等。在Java NIO中同步事件分离器对应的组件就是Selector对应的阻塞方法就是select方法。Event Handler(事件处理器) 本身由多个回调方法构成这些回调构成了与应用相关的对于某个事件的反馈机制。Netty相比于Java NIO来说在事件处理器这个角色上进行一个升级它为我们开发者提供了大量的回调方法供我们在待定事件产生时实现相应的回调方法进行业务逻辑的处理。Concrete Event Handler(具体事件处理器)它本身实现了事件处理所提供的各个回调方法从而实现了特定于业务的逻辑。它本质上就是我们所编写的一个个的处理器实现。Initiation Dispatcher(初始分发器)实际上就是Reactor角色。它本身定义了一些规范这些规范用于控制事件的调度方式同时又提供了应用进行事件处理器的注册、删除等。Initiation Dispatcher会通过同步事件分离器来等待事件的发生一旦事件发生Initiation Dispatcher首先会分离出每一个事件然后调用事件处理器最后调用相关的回调方法来处理事件。执行流程分析当应用像Initiation Dispatcher注册具体的事件处理器时应用会标识出事件处理器希望Initiation Dispatcher在某个事件发生时向其通知该事件该事件与Handle关联。Initiation Dispatcher会要求每个事件向其传递内部的Handle。该Handle向操作系统标识了事件处理器。当所有事件处理器注册完毕后应用会调用handle_events方法来启动Initiation Dispatcher的事件循环。这时Initiation Dispatcher会将每个注册的事件管理器的Handle合并起来并使用同步事件分离器等待这些事件的发生。比如说TCP协议层使用select同步事件分离器操作来等待客户端发送的数据到达连接的socker handle上。当与某个事件源对应的Handle变为ready状态时(比如说TCP socker变为等待读状态时)同步事件分离器就会通知Initiation Dispatcher。Initiation Dispatcher会触发事件处理器的回调方法从而响应这个处于ready状态的Handle。Initiation Dispatcher会回调事件处理器的handle_events回调方法来执行特定于应用的功能(开发者自己所编写的功能)从而响应这个事件。所发生的事件类型可以作为该方法参数并被该方法内部使用来执行额外的特定于服务的功能。以上描述的内容似乎和本文的标题不大其实不然它正是下面介绍的内容的开端。2. 通过一个例子拉近与Java NIO的距离/*** Author CoderJiA* Description NIOServer* Date 13/2/19 下午4:59**/public class NIOServer {public static void main(String[] args) throws Exception{// 1.创建ServerSocketChannelServerSocketChannel serverSocketChannel ServerSocketChannel.open();serverSocketChannel.configureBlocking(false);ServerSocket serverSocket serverSocketChannel.socket();serverSocket.bind(new InetSocketAddress(8899));// 2.创建Selector并ServerSocketChannel注册OP_ACCEPT事件接收连接。Selector selector Selector.open();serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);// 3.开启轮询while (selector.select() 0) {// 从selector所有事件就绪的key并遍历处理。Set selectionKeys selector.selectedKeys();selectionKeys.forEach(selectionKey - {SocketChannel client;try {if (selectionKey.isAcceptable()) { // 接受事件就绪// 获取serverSocketChannelServerSocketChannel server (ServerSocketChannel)selectionKey.channel();// 接收连接client server.accept();client.configureBlocking(false);client.register(selector, SelectionKey.OP_READ);} else if (selectionKey.isReadable()) { // 读事件就绪// 获取socketChannelclient (SocketChannel) selectionKey.channel();// 创建buffer,并将获取socketChannel中的数据读入到buffer中ByteBuffer readBuf ByteBuffer.allocate(1024);int readCount client.read(readBuf);if (readCount 0) {return;}Charset charset Charset.forName(StandardCharsets.UTF_8.name());readBuf.flip();System.out.println(String.valueOf(charset.decode(readBuf).array()));}} catch (IOException e) {e.printStackTrace();}selectionKeys.remove(selectionKey);});}}复制代码通过这个例子与原始Reactor模式相对应的理解比如同步事件分离器对应着Selector的select()方法再比如ServerSocketChannel注册给Selector的OP_ACCEPT还有SocketChannel的OP_READ与OP_WRITE这些事件保存在操作系统上其实就是原始Reactor中的Handle。四个重要apiChannelConnections to files,sockets etc that support non-blocking reads.BufferArray-like objects that can be directly read or written by Channels.SelectorTell which of a set of Channels have IO events.SelectionKeysMaintain IO event status and bingdings.3.用Java NIO对Reactor模式的应用。3.1 Single threaded version/*** Author CoderJiA* Description Reactor* Date 5/4/19 下午2:25**/public abstract class Reactor implements Runnable{protected final Selector selector;protected final ServerSocketChannel serverSocket;protected final long port;protected final long timeout;public Reactor(int port, long timeout) throws IOException {this.port port;this.timeout timeout;selector Selector.open();serverSocket ServerSocketChannel.open();serverSocket.socket().bind(new InetSocketAddress(port));serverSocket.configureBlocking(false);SelectionKey sk serverSocket.register(selector, SelectionKey.OP_ACCEPT);sk.attach(newAcceptor(selector));}Overridepublic void run() {try {while (!Thread.interrupted()) {if (selector.select(timeout) 0) {Set selected selector.selectedKeys();selected.forEach(sk - {dispatch(sk);selected.remove(sk);});}}} catch (IOException e) {e.printStackTrace();}}private void dispatch(SelectionKey sk) {Runnable r (Runnable)(sk.attachment());if (Objects.nonNull(r)) {r.run();}}public abstract Acceptor newAcceptor(Selector selector);}复制代码/*** Author CoderJiA* Description Acceptor* Date 5/4/19 下午2:58**/public class Acceptor implements Runnable {private final Selector selector;private final ServerSocketChannel serverSocket;public Acceptor(Selector selector, ServerSocketChannel serverSocket) {this.selector selector;this.serverSocket serverSocket;}Overridepublic void run() {try {SocketChannel socket serverSocket.accept();if (Objects.nonNull(socket)) {new Handler(selector, socket);}} catch (IOException e) {e.printStackTrace();}}}复制代码/*** Author CoderJiA* Description Handler* Date 5/4/19 下午4:25**/public class Handler implements Runnable {private static final int MB 1024 * 1024;protected final SocketChannel socket;protected final SelectionKey sk;protected final ByteBuffer input ByteBuffer.allocate(MB);protected final ByteBuffer output ByteBuffer.allocate(MB);private static final int READING 0, SENDING 1;private int state READING;public Handler(Selector selector, SocketChannel socket) throws IOException {this.socket socket;socket.configureBlocking(false);sk socket.register(selector, SelectionKey.OP_READ);sk.attach(this);}Overridepublic void run() {try {if (state READING) read();else if (state SENDING) send();} catch (Exception e) {e.printStackTrace();}}private void read() throws IOException {socket.read(input);if (inputIsComplete()) {state SENDING;sk.interestOps(SelectionKey.OP_WRITE);}input.clear();}private void send() throws IOException {socket.write(output);if (outputIsComplete()) {sk.cancel();}}private boolean inputIsComplete() {return input.position() 0;}private boolean outputIsComplete() {return !output.hasRemaining();}}复制代码/*** Author CoderJiA* Description EchoReactor* Date 5/4/19 下午5:01**/public class EchoReactor extends Reactor {private static final int PORT 9999;private static final long TIME_OUT TimeUnit.MILLISECONDS.toMillis(10);public EchoReactor(int port, long timeout) throws IOException {super(port, timeout);}Overridepublic Acceptor newAcceptor(Selector selector) {return new Acceptor(selector, this.serverSocket);}public static void main(String[] args) throws IOException {new EchoReactor(PORT, TIME_OUT).run();}}复制代码核心组件组件分析Reactor等同于原始Reactor模式的Initiation Dispatcher它负责所有就绪事件统一分发到事件处理器如Acceptor和Hanlder。Acceptor用于将接收到的SocketChannel交给Handler处理。Handler处理读写操作。这是Reactor的单线程版本这个版本一个线程处理客户端的接收和数据处理以及读写操作数据处理往往就是我们实际开发中的业务处理是比较耗时的。如果一个处理过程处于阻塞那么这个模型所表现出的就处于阻塞所以一个数据处理的阻塞会导致不能处理客户端连接的接收。因此衍生出来下面的多工作线程版本来优化Handler。3.2 Worker Threads version调整下Handlerpackage cn.coderjia.nio.douglea.reactor2;import java.io.IOException;import java.nio.ByteBuffer;import java.nio.channels.SelectionKey;import java.nio.channels.Selector;import java.nio.channels.SocketChannel;import java.util.concurrent.ExecutorService;import java.util.concurrent.Executors;/*** Author CoderJiA* Description Handler* Date 5/4/19 下午4:25**/public class Handler implements Runnable {private static final int MB 1024 * 1024;protected final SocketChannel socket;protected final SelectionKey sk;protected final ByteBuffer input ByteBuffer.allocate(MB);protected final ByteBuffer output ByteBuffer.allocate(MB);private static final int READING 0, SENDING 1, PROCESSING 3;private int state READING;private static final ExecutorService EXECUTOR_SERVICE Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors());public Handler(Selector selector, SocketChannel socket) throws IOException {this.socket socket;socket.configureBlocking(false);sk socket.register(selector, SelectionKey.OP_READ);sk.attach(this);}Overridepublic void run() {try {if (state READING) read();else if (state SENDING) send();} catch (Exception e) {e.printStackTrace();}}private void read() throws IOException {socket.read(input);if (inputIsComplete()) {state PROCESSING;EXECUTOR_SERVICE.execute(new Processer());}input.clear();}private void send() throws IOException {socket.write(output);if (outputIsComplete()) {sk.cancel();}}private void process() {System.out.println(Handler.process()...);}private boolean inputIsComplete() {return input.position() 0;}private boolean outputIsComplete() {return !output.hasRemaining();}class Processer implements Runnable {public void run() {processAndHandOff();}}synchronized void processAndHandOff() {process();state SENDING;sk.interestOps(SelectionKey.OP_WRITE);}}复制代码Handler多工作线程版本将耗时的process()创建线程去处理。这个版本Reactor既负责客户端的接收事件又负责读写事件因为对于高并发场景连接数巨大Reactor可能有时候会力不从心。因此衍生出下面的主从Reactor模型。3.3 Multiple Reactors Version调整Acceptor/*** Author CoderJiA* Description Acceptor3* Date 6/4/19 下午6:51**/public class Acceptor3 implements Runnable {private final ServerSocketChannel serverSocket;public Acceptor3(ServerSocketChannel serverSocket) {this.serverSocket serverSocket;}Overridepublic void run() {try {SocketChannel socket serverSocket.accept();if (Objects.nonNull(socket)) {new Handler(EchoReactor.nextSubReactor().selector, socket);}} catch (IOException e) {e.printStackTrace();}}}复制代码调整Reactor/*** Author CoderJiA* Description Reactor3* Date 6/4/19 下午6:51**/public abstract class Reactor3 implements Runnable {protected Selector selector;protected ServerSocketChannel serverSocket;protected final int port;protected final long timeout;protected final boolean isMainReactor;public Reactor3(int port, long timeout, boolean isMainReactor) {this.port port;this.timeout timeout;this.isMainReactor isMainReactor;}Overridepublic void run() {try {init();while (!Thread.interrupted()) {if (selector.select(timeout) 0) {System.out.println(isMainReactor: isMainReactor);Set selected selector.selectedKeys();selected.forEach(sk - {dispatch(sk);selected.remove(sk);});selected.clear();}}} catch (IOException e) {e.printStackTrace();}}private void init() throws IOException {selector Selector.open();if (isMainReactor) {serverSocket ServerSocketChannel.open();serverSocket.socket().bind(new InetSocketAddress(port));serverSocket.configureBlocking(false);SelectionKey sk serverSocket.register(selector, SelectionKey.OP_ACCEPT);sk.attach(newAcceptor());}}private void dispatch(SelectionKey sk) {Runnable r (Runnable)(sk.attachment());if (Objects.nonNull(r)) {r.run();}}public abstract Acceptor3 newAcceptor();}复制代码/*** Author CoderJiA* Description EchoReactor* Date 6/4/19 下午5:35**/public class EchoReactor extends Reactor3 {private static final int PORT 9999;private static final long TIME_OUT TimeUnit.MILLISECONDS.toMillis(10);private static final int SUB_REACTORS_SIZE 2;private static final Reactor3[] SUB_REACTORS new Reactor3[SUB_REACTORS_SIZE];private static final AtomicInteger NEXT_INDEX new AtomicInteger(0);static {// 初始化子ReactorIntStream.range(0, SUB_REACTORS_SIZE).forEach(i - SUB_REACTORS[i] new EchoReactor(PORT, TIME_OUT, false));}public static Reactor3 nextSubReactor(){int curIdx NEXT_INDEX.getAndIncrement();if(curIdx SUB_REACTORS_SIZE){NEXT_INDEX.set(0);curIdx 0;}return SUB_REACTORS[(curIdx % SUB_REACTORS_SIZE)];}public EchoReactor(int port, long timeout, boolean isMainReactor) {super(port, timeout, isMainReactor);}Overridepublic Acceptor3 newAcceptor() {return new Acceptor3(this.serverSocket);}public static void main(String[] args) {Reactor3 mainReactor new EchoReactor(PORT, TIME_OUT, true);// 启动主Reactornew Thread(mainReactor).start();// 启动子ReactorIntStream.range(0, SUB_REACTORS_SIZE).forEach(i - new Thread(SUB_REACTORS[i]).start());}}复制代码主从Reactor模型主Reactor用于处理客户端连接的接收转发给Acceptor处理子Reactor处理读写事件的接收转发给Handler处理。参考文章Scalable IO in Java源码地址
