多线程 reactor 模型

Author: haobinaa

README.md

@@ -67,6 +67,7 @@
  、- [缓存击穿解决方法](/src/main/java/com/haobin/codeBlock/CacheBreakdown.java)
  - [传统IO模型](/src/main/java/com/haobin/codeBlock/IOModel/ClassicServerLoop.java) 
  - [IO-reactor模式(单线程)](/src/main/java/com/haobin/codeBlock/IOModel/SingleReactor.java)
+  - [IO-reactor模式(多线程)](/src/main/java/com/haobin/codeBlock/IOModel/MultiReactor.java)
 
 
 

src/main/java/com/haobin/codeBlock/IOModel/BasicHandler.java

@@ -0,0 +1,160 @@
+package com.haobin.codeBlock.IOModel;
+
+import java.io.EOFException;
+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.nio.charset.StandardCharsets;
+
+/**
+ * @author: HaoBin
+ * @create: 2019/10/23 15:16
+ * @description: 单线程reactor handler 处理器
+ * <p>
+ * 在单线程reactor中， IO读写以及业务处理均由reactor线程完成
+ **/
+public class BasicHandler implements Runnable {
+
+    private static final int INPUT_SIZE = 1024;
+    private static final int OUTPUT_SIZE = 1024;
+
+    public SocketChannel socket;
+    public SelectionKey selectionKey;
+    ByteBuffer input = ByteBuffer.allocate(INPUT_SIZE);
+    ByteBuffer output = ByteBuffer.allocate(OUTPUT_SIZE);
+
+    // 定义服务的逻辑状态
+    static final int READING = 0;
+    static final int SENDING = 1;
+    static final int CLOSED = 2;
+    int state = READING;
+
+
+    public BasicHandler(SocketChannel socket) {
+        this.socket = socket;
+    }
+
+    public BasicHandler(Selector selector, SocketChannel sc) throws IOException {
+        socket = sc;
+        sc.configureBlocking(false); // 非阻塞
+        // 注册到selector上，但不订阅任何事件
+        selectionKey = socket.register(selector, 0);
+        // 绑定要处理事件
+        selectionKey.interestOps(SelectionKey.OP_READ);
+        // 处理事件对应的程序
+        selectionKey.attach(this);
+        // 唤醒selector
+        selector.wakeup();
+    }
+
+    @Override
+    public void run() {
+        try {
+            if (state == READING)
+                read(); // 此时通道已经准备好读取字节
+            else if (state == SENDING)
+                send(); // 此时通道已经准备好写入字节
+        } catch (IOException ex) {
+            // 关闭连接
+            try {
+                selectionKey.channel().close();
+            } catch (IOException ignore) {
+            }
+        }
+    }
+
+    protected void read() throws IOException {
+        input.clear(); // 清空接收缓冲区
+        int n = socket.read(input);
+        if (inputIsComplete(n)) {// 如果读取了完整的数据
+            process();
+            // 待发送的数据已经放入发送缓冲区中
+
+            // 更改服务的逻辑状态以及要处理的事件类型
+            selectionKey.interestOps(SelectionKey.OP_WRITE);
+        }
+    }
+
+    // 缓存每次读取的内容
+    StringBuilder request = new StringBuilder();
+    /**
+     * 当读取到 \r\n 时表示结束
+     * @param bytes 读取的字节数，-1 通常是连接被关闭，0 非阻塞模式可能返回
+     * @throws IOException
+     */
+    protected boolean inputIsComplete(int bytes) throws IOException {
+        if (bytes > 0) {
+            input.flip(); // 切换成读取模式
+            while (input.hasRemaining()) {
+                byte ch = input.get();
+
+                if (ch == 3) { // ctrl+c 关闭连接
+                    state = CLOSED;
+                    return true;
+                } else if (ch == '\r') { // continue
+                } else if (ch == '\n') {
+                    // 读取到了 \r\n 读取结束
+                    state = SENDING;
+                    return true;
+                } else {
+                    request.append((char)ch);
+                }
+            }
+        } else if (bytes == -1) {
+            // -1 客户端关闭了连接
+            throw new EOFException();
+        } else {} // bytes == 0 继续读取
+        return false;
+    }
+
+    /**
+     * 根据业务处理结果，判断如何响应
+     * @throws EOFException 用户输入 ctrl+c 主动关闭
+     */
+    protected void process() throws EOFException {
+        if (state == CLOSED) {
+            throw new EOFException();
+        } else if (state == SENDING) {
+            String requestContent = request.toString(); // 请求内容
+            byte[] response = requestContent.getBytes(StandardCharsets.UTF_8);
+            output.put(response);
+        }
+    }
+
+    protected void send() throws IOException {
+        int written = -1;
+        output.flip();// 切换到读取模式，判断是否有数据要发送
+        if (output.hasRemaining()) {
+            written = socket.write(output);
+        }
+
+        // 检查连接是否处理完毕，是否断开连接
+        if (outputIsComplete(written)) {
+            selectionKey.channel().close();
+        } else {
+            // 否则继续读取
+            state = READING;
+            // 把提示发到界面
+            socket.write(ByteBuffer.wrap("\r\nreactor> ".getBytes()));
+            selectionKey.interestOps(SelectionKey.OP_READ);
+        }
+
+    }
+
+    /**
+     * 当用户输入了一个空行，表示连接可以关闭了
+     */
+    protected boolean outputIsComplete(int written) {
+        if (written <= 0) {
+            // 用户只敲了个回车， 断开连接
+            return true;
+        }
+
+        // 清空旧数据，接着处理后续的请求
+        output.clear();
+        request.delete(0, request.length());
+        return false;
+    }
+}

src/main/java/com/haobin/codeBlock/IOModel/MultiHandler.java

@@ -0,0 +1,90 @@
+package com.haobin.codeBlock.IOModel;
+
+import java.io.EOFException;
+import java.io.IOException;
+import java.nio.channels.SelectionKey;
+import java.nio.channels.Selector;
+import java.nio.channels.SocketChannel;
+import java.util.concurrent.ArrayBlockingQueue;
+import java.util.concurrent.ThreadFactory;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+
+/**
+ * @author: HaoBin
+ * @create: 2019/10/23 16:11
+ * @description: 多线程 reactor handler
+ **/
+public class MultiHandler extends BasicHandler {
+
+    static ThreadPoolExecutor workPool =
+            new ThreadPoolExecutor(1,
+                    1,
+                    60, TimeUnit.SECONDS,
+                    new ArrayBlockingQueue<>(10),
+                    new ThreadFactory() {
+                        private AtomicInteger count = new AtomicInteger(0);
+                        @Override
+                        public Thread newThread(Runnable r) {
+                            Thread t = new Thread(r);
+                            t.setName(MultiHandler.class.getName() + count.addAndGet(1));
+                            return t;
+                        }
+                    },
+                    new ThreadPoolExecutor.AbortPolicy());
+
+    static final int PROCESSING = 4;
+    private Object lock = new Object();
+
+    public MultiHandler(Selector selector, SocketChannel sc) throws IOException {
+        super(selector, sc);
+    }
+
+    @Override
+    protected void read() throws IOException {
+        // 为什么要同步？Processer 线程处理时通道还有可能有读事件发生
+        // 保护 input ByteBuffer 不会重置和状态的可见性
+        // 应该是这样
+        synchronized (lock) {
+            input.clear();
+            int n = socket.read(input);
+            if (inputIsComplete(n)) {
+
+                // 读取完毕后将后续的处理交给
+                state = PROCESSING;
+                workPool.execute(new Processer());
+            }
+        }
+    }
+
+    class Processer implements Runnable {
+        @Override
+        public void run() {
+            processAndHandOff();
+        }
+    }
+
+    private void processAndHandOff() {
+        synchronized (lock) {
+            try {
+                process();
+            } catch (EOFException e) {
+                // 直接关闭连接
+                try {
+                    selectionKey.channel().close();
+                } catch (IOException e1) {}
+                return;
+            }
+
+            // 最后的发送还是交给 SingleReactor 线程处理
+            state = SENDING;
+            selectionKey.interestOps(SelectionKey.OP_WRITE);
+
+            // 这里需要唤醒 Selector，因为当把处理交给 workpool 时，SingleReactor 线程已经阻塞在 select() 方法了， 注意
+            // 此时该通道感兴趣的事件还是 OP_READ，这里将通道感兴趣的事件改为 OP_WRITE，如果不唤醒的话，就只能在
+            // 下次select 返回时才能有响应了，当然了也可以在 select 方法上设置超时
+            selectionKey.selector().wakeup();
+        }
+    }
+}