Reactor之多工作线程与主从模式

阳光穿透心脏的1/2处 2022-12-28 14:13 177阅读 0赞

# 多线程、主从Reactor #

## 多工作线程模式 ##

单线程的reactor模式并没有解决IO和CPU处理速度不匹配问题，所以多线程的reactor模式引入线程池的概念，把耗时的IO操作交由线程池处理。

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMjI4MTI4NDk_size_16_color_FFFFFF_t_70_pic_center]

package com.morris.reactor.multiworker;
    
    import java.io.IOException;
    import java.net.InetSocketAddress;
    import java.nio.channels.SelectionKey;
    import java.nio.channels.Selector;
    import java.nio.channels.ServerSocketChannel;
    import java.util.Set;
    
    /** * 单线程Reactor模式，具体业务逻辑交由线程池处理 */
    public class Reactor implements Runnable { 
    
        private Selector selector;
    
        public Reactor(int port) throws IOException { 
            selector = Selector.open();
            ServerSocketChannel ssc = ServerSocketChannel.open();
            ssc.bind(new InetSocketAddress(port));
            ssc.configureBlocking(false);
            ssc.register(selector, SelectionKey.OP_ACCEPT, new Acceptor(ssc, selector));
        }
    
        @Override
        public void run() { 
            while (!Thread.interrupted()) { 
                try { 
                    selector.select();
                    Set<SelectionKey> selectionKeys = selector.selectedKeys();
                    for (SelectionKey selectionKey : selectionKeys) { 
    
                        Runnable r = (Runnable) selectionKey.attachment();
                        if(null != r) { 
                            r.run();
                        }
                    }
                    selectionKeys.clear();
                } catch (IOException e) { 
                    e.printStackTrace();
                }
            }
        }
    }
    
    package com.morris.reactor.multiworker;
    
    import java.io.IOException;
    import java.nio.channels.SelectionKey;
    import java.nio.channels.Selector;
    import java.nio.channels.ServerSocketChannel;
    import java.nio.channels.SocketChannel;
    
    public class Acceptor implements Runnable { 
    
        private ServerSocketChannel ssc;
    
        private Selector selector;
    
        public Acceptor(ServerSocketChannel ssc, Selector selector) { 
            this.ssc = ssc;
            this.selector = selector;
        }
    
        @Override
        public void run() { 
            try { 
                SocketChannel sc = ssc.accept();
                sc.configureBlocking(false);
                sc.register(selector, SelectionKey.OP_READ, new Handler(sc, selector));
            } catch (IOException e) { 
                e.printStackTrace();
            }
        }
    }
    
    package com.morris.reactor.multiworker;
    
    
    import lombok.extern.slf4j.Slf4j;
    
    import java.io.IOException;
    import java.nio.ByteBuffer;
    import java.nio.channels.ClosedChannelException;
    import java.nio.channels.SelectionKey;
    import java.nio.channels.Selector;
    import java.nio.channels.SocketChannel;
    import java.time.temporal.ValueRange;
    import java.util.concurrent.ExecutorService;
    import java.util.concurrent.Executors;
    
    @Slf4j
    public class Handler implements Runnable { 
    
        private SocketChannel sc;
        private Selector selector;
    
        ByteBuffer in = ByteBuffer.allocate(1024);
        ByteBuffer out = ByteBuffer.allocate(1024);
    
        ExecutorService executorService = Executors.newFixedThreadPool(3);
    
        public Handler(SocketChannel sc, Selector selector) { 
            this.sc = sc;
            this.selector = selector;
        }
    
        @Override
        public void run() { 
    
            // 第一次进来肯定是读事件
            try { 
                read();
    
                // 业务逻辑交给业务线程池处理
                executorService.submit(new Processor());
    
                write();
            } catch (IOException e) { 
                e.printStackTrace();
            }
        }
    
        private void read() throws IOException { 
            // 1. read
            sc.read(in); // read from socket channel to byte buffer
            in.flip();
        }
    
        private void process() throws ClosedChannelException { 
            byte[] bytes = new byte[in.remaining()];
            in.get(bytes); // read from byte buffer to byte array
            // 2. decode
            String request = new String(bytes);
    
            // 3. process
            log.info("receive from client: {}", request);
    
            // 4. decode
            byte[] responseBytes = "hi".getBytes();
            out.put(responseBytes); // write byte array to byte buffer
            out.flip();
    
            sc.register(selector, SelectionKey.OP_WRITE, new Sender());
        }
    
        private void write() throws IOException { 
            // 5. write
            sc.write(out); // write byte buffer to socket channel
            sc.close();
        }
    
        private class Sender implements Runnable { 
    
            @Override
            public void run() { 
                try { 
                    write();
                } catch (IOException e) { 
                    e.printStackTrace();
                }
            }
        }
    
        private class Processor implements Runnable { 
    
            @Override
            public void run() { 
                try { 
                    process();
                } catch (IOException e) { 
                    e.printStackTrace();
                }
            }
        }
    
    }

总的来说，多线程版本的reactor是为了解决单线程reactor版本的IO和CPU处理速度不匹配问题，从而达到高效处理的目的。

## 主从Reactor模式 ##

主Reactor负责接受连接，使用多个从reactor进行接收数据和发送数据。

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMjI4MTI4NDk_size_16_color_FFFFFF_t_70_pic_center 1]

package com.morris.reactor.multireactor;
    
    import java.io.IOException;
    import java.net.InetSocketAddress;
    import java.nio.channels.SelectionKey;
    import java.nio.channels.Selector;
    import java.nio.channels.ServerSocketChannel;
    import java.util.Set;
    
    /** * 一主多从Reactor模式 */
    public class MainReactor implements Runnable { 
    
        private Selector selector;
    
        public MainReactor(int port) throws IOException { 
            selector = Selector.open();
            ServerSocketChannel ssc = ServerSocketChannel.open();
            ssc.bind(new InetSocketAddress(port));
            ssc.configureBlocking(false);
            ssc.register(selector, SelectionKey.OP_ACCEPT, new Acceptor(ssc));
        }
    
        @Override
        public void run() { 
            while (!Thread.interrupted()) { 
                try { 
                    selector.select();
                    Set<SelectionKey> selectionKeys = selector.selectedKeys();
                    for (SelectionKey selectionKey : selectionKeys) { 
    
                        Runnable r = (Runnable) selectionKey.attachment();
                        if(null != r) { 
                            r.run();
                        }
                    }
                    selectionKeys.clear();
                } catch (IOException e) { 
                    e.printStackTrace();
                }
            }
        }
    }
    
    package com.morris.reactor.multireactor;
    
    import java.io.IOException;
    import java.nio.channels.SelectionKey;
    import java.nio.channels.Selector;
    import java.nio.channels.ServerSocketChannel;
    import java.nio.channels.SocketChannel;
    
    public class Acceptor implements Runnable { 
    
        public static final int MAX_THREAD_NUM = 3;
    
        private ServerSocketChannel ssc;
    
        private Selector[] selectors;
        private Thread[] threads;
        private int next;
    
        public Acceptor(ServerSocketChannel ssc) throws IOException { 
            this.ssc = ssc;
            selectors = new Selector[MAX_THREAD_NUM];
            threads = new Thread[MAX_THREAD_NUM];
            for (int i = 0; i < MAX_THREAD_NUM; i++) { 
                selectors[i] = Selector.open();
                threads[i] = new Thread(new SubReactor(selectors[i]), "SubReactor-" + i);
                threads[i].start();
            }
        }
    
        @Override
        public void run() { 
            try { 
                SocketChannel sc = ssc.accept();
                sc.configureBlocking(false);
    // selectors[next].wakeup();
                sc.register(selectors[next], SelectionKey.OP_READ, new Handler(sc, selectors[next]));
                if (++next == selectors.length) { 
                    next = 0;
                }
            } catch (IOException e) { 
                e.printStackTrace();
            }
        }
    }
    
    package com.morris.reactor.multireactor;
    
    import java.io.IOException;
    import java.net.InetSocketAddress;
    import java.nio.channels.SelectionKey;
    import java.nio.channels.Selector;
    import java.nio.channels.ServerSocketChannel;
    import java.util.Set;
    
    /** * 单线程Reactor模式 */
    public class SubReactor implements Runnable { 
    
        private Selector selector;
    
        public SubReactor(Selector selector) { 
            this.selector = selector;
        }
    
        @Override
        public void run() { 
            while (!Thread.interrupted()) { 
                try { 
                    // selector.select();
                    selector.selectNow();
                    Set<SelectionKey> selectionKeys = selector.selectedKeys();
                    for (SelectionKey selectionKey : selectionKeys) { 
    
                        Runnable r = (Runnable) selectionKey.attachment();
                        if(null != r) { 
                            r.run();
                        }
                    }
                    selectionKeys.clear();
                } catch (IOException e) { 
                    e.printStackTrace();
                }
            }
        }
    }
    
    package com.morris.reactor.multireactor;
    
    
    import lombok.extern.slf4j.Slf4j;
    
    import java.io.IOException;
    import java.nio.ByteBuffer;
    import java.nio.channels.ClosedChannelException;
    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;
    
    @Slf4j
    public class Handler implements Runnable { 
    
        private SocketChannel sc;
        private Selector selector;
    
        ByteBuffer in = ByteBuffer.allocate(1024);
        ByteBuffer out = ByteBuffer.allocate(1024);
    
        ExecutorService executorService = Executors.newFixedThreadPool(3);
    
        public Handler(SocketChannel sc, Selector selector) { 
            this.sc = sc;
            this.selector = selector;
        }
    
        @Override
        public void run() { 
    
            // 第一次进来肯定是读事件
            try { 
                read();
    
                // 业务逻辑交给业务线程池处理
                executorService.submit(new Processor());
    
                write();
            } catch (IOException e) { 
                e.printStackTrace();
            }
        }
    
        private void read() throws IOException { 
            // 1. read
            sc.read(in); // read from socket channel to byte buffer
            in.flip();
        }
    
        private void process() throws ClosedChannelException { 
            byte[] bytes = new byte[in.remaining()];
            in.get(bytes); // read from byte buffer to byte array
            // 2. decode
            String request = new String(bytes);
    
            // 3. process
            log.info("receive from client: {}", request);
    
            // 4. decode
            byte[] responseBytes = "hi".getBytes();
            out.put(responseBytes); // write byte array to byte buffer
            out.flip();
    
            sc.register(selector, SelectionKey.OP_WRITE, new Sender());
        }
    
        private void write() throws IOException { 
            // 5. write
            sc.write(out); // write byte buffer to socket channel
            sc.close();
        }
    
        private class Sender implements Runnable { 
    
            @Override
            public void run() { 
                try { 
                    write();
                } catch (IOException e) { 
                    e.printStackTrace();
                }
            }
        }
    
        private class Processor implements Runnable { 
    
            @Override
            public void run() { 
                try { 
                    process();
                } catch (IOException e) { 
                    e.printStackTrace();
                }
            }
        }
    
    }

上面的代码如果不注意的话，会产生死锁，下面进行死锁原因的分析：

1.  SubReactor.run()方法中会调用`selector.select()`，这段代码会一直阻塞，直至有数据到来，但是此时Selector中并没有注册任何SocketChannel，所以永远也不会有数据到来，会一直阻塞。
2.  当Acceptor发送新的连接到来时，会将新到来的SocketChannel注册到SubReactor的Selector中，会调用`sc.register()`方法，此时就会产生死锁。

selector.select()的关键部分源码：

private int lockAndDoSelect(long var1) throws IOException { 
        synchronized(this) { 
            if (!this.isOpen()) { 
                throw new ClosedSelectorException();
            } else { 
                int var10000;
                synchronized(this.publicKeys) { 
                    synchronized(this.publicSelectedKeys) { 
                        var10000 = this.doSelect(var1); // 阻塞在这里，一直持有publicKeys锁
                    }
                }
    
                return var10000;
            }
        }
    }

sc.register()的关键部分源码：

protected final SelectionKey register(AbstractSelectableChannel var1, int var2, Object var3) { 
        if (!(var1 instanceof SelChImpl)) { 
            throw new IllegalSelectorException();
        } else { 
            SelectionKeyImpl var4 = new SelectionKeyImpl((SelChImpl)var1, this);
            var4.attach(var3);
            synchronized(this.publicKeys) {  // 这里会获取不到，产生死锁
                this.implRegister(var4);
            }
    
            var4.interestOps(var2);
            return var4;
        }
    }

解决死锁的办法：

1.  使用selector.selectNow()这种非阻塞的方法，或者selector.select(time)带超时时间的方法。
2.  借助selectors.wakeup()，这个方法可以唤醒`selector.select()`。比如可以为SubReactor设置一个标志位flag，如果flag=true，可以执行selector.select()，否则就会阻塞等待，当需要往selector上注册新的SocketChannel时，先将flag设置为false，然后selectors.wakeup()唤醒select()，然后sc.register()注册Channel，最后还原flag。

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMjI4MTI4NDk_size_16_color_FFFFFF_t_70_pic_center]: /images/20221120/3b05c7874a564fabbaefc39608160e7e.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMjI4MTI4NDk_size_16_color_FFFFFF_t_70_pic_center 1]: /images/20221120/33535415eb1743c883f770716b09f04c.png