Reactor模型示例

ゞ浴缸里的玫瑰 2023-02-17 05:27 9阅读 0赞

Reactor模型相关内容可参考：[https://blog.csdn.net/skh2015java/article/details/106721630][https_blog.csdn.net_skh2015java_article_details_106721630]

Reactor模型示例
    mainSelector 负责处理ACCEPT I/O事件，处理ACCEPT的业务逻辑Handler交给executorService线程池处理
    subSelector负责处理READ I/O事件，处理READ和逻辑的Handler交给executorService线程处理

import java.io.IOException;
    import java.net.InetSocketAddress;
    import java.nio.ByteBuffer;
    import java.nio.channels.*;
    import java.nio.charset.Charset;
    import java.util.Iterator;
    import java.util.concurrent.ConcurrentLinkedQueue;
    import java.util.concurrent.ExecutorService;
    import java.util.concurrent.Executors;
    
    
    
    public class ReactorTask {
    
    
        public static void main(String[] args) throws IOException {
            new ReactorTask().startReactor();
        }
    
    
    
        final int  PORT = 7786;
        final Selector  mainSelector;
        final Selector subSelector;
        final ServerSocketChannel serverSocketChannel;
    
        final Thread  subThread ;
        final ExecutorService executorService = Executors.newFixedThreadPool(8);
    
        private ConcurrentLinkedQueue <SocketChannel> socketChannels = new ConcurrentLinkedQueue<>();
    
        public ReactorTask() throws IOException {
            mainSelector = Selector.open();
            subSelector = Selector.open();
    
            subThread = new Thread(new SubThreadTask());
    
            serverSocketChannel = ServerSocketChannel.open();
            serverSocketChannel.configureBlocking(false);
            serverSocketChannel.bind(new InetSocketAddress(PORT));
    
            serverSocketChannel.register(mainSelector, SelectionKey.OP_ACCEPT);
        }
    
        public void startReactor(){
            try{
                subThread.start();
                while (!Thread.interrupted()){
                    mainSelector.select();  //阻塞等待ACCEPT事件
                    Iterator<SelectionKey> it = mainSelector.selectedKeys().iterator();
                    while (it.hasNext()) {
                        SelectionKey key=  it.next();
                        if (key.isAcceptable()) {
                            ServerSocketChannel server = (ServerSocketChannel)key.channel();
                            SocketChannel sc = server.accept();
                            executorService.execute(new ServerSocketChannelAcceptHandle(sc));
                        }
                    }
                }
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    
    
    
        class SubThreadTask implements Runnable {
            @Override
            public void run() {
                while (!Thread.currentThread().isInterrupted()) {
                    try {
                           subSelector.select();
                            while (!socketChannels.isEmpty()){
                                SocketChannel socketChannel  = socketChannels.poll();
                                socketChannel.configureBlocking(false);
                                socketChannel.register(subSelector,SelectionKey.OP_READ);
                                ByteBuffer buffer = ByteBuffer.wrap("register success".getBytes());
                                socketChannel.write(buffer);
                            }
    
    
                        Iterator<SelectionKey> selectedKeys = subSelector.selectedKeys().iterator();
                        while (selectedKeys.hasNext()){
                            SelectionKey key =  selectedKeys.next();
                            if (!key.isValid()) {
                                break;
                            }
                            if (key.isReadable()){
                                executorService.execute(new SocketChannelReadHandle(key));
                            }
                        }
                    } catch (Exception e) {
                        System.out.println("catch Exception "+ e);
                    }
                }
            }
        }
    
    
        class SocketChannelReadHandle implements Runnable {
    
            private SelectionKey key;
            ByteBuffer buffer = ByteBuffer.allocate(128);
    
            public SocketChannelReadHandle(SelectionKey key){
                this.key = key;
            }
    
            @Override
            public void run() {
                try {
                    if (!key.isValid()){
                        return;
                    }
                    SocketChannel channel = (SocketChannel)key.channel();
                    buffer.clear();
                    if ( channel.read(buffer) > 0 ){
                        buffer.flip();
                        System.out.println("handle Thread:"+Thread.currentThread().getName() +"  subReactor  read:"+Charset.forName("UTF-8").decode(buffer).toString() );
                        buffer.clear();
                        buffer.put("I am subReactor".getBytes());
                        buffer.flip();
                        channel.write(buffer);
                    }
                } catch (Exception e) {
                    System.out.println(" caught "+ e + " closing channel");
                    try {
                            key.channel().close();
                    } catch (Exception e1) {
                        System.out.println("caught" + e1 + "channel has closed ");
                    }
                }
            }
        }
    
    
        class ServerSocketChannelAcceptHandle implements Runnable {
            private SocketChannel channel;
    
            public ServerSocketChannelAcceptHandle(SocketChannel channel){
                this.channel = channel;
            }
    
    
            @Override
            public void run() {
                try {
                    socketChannels.offer(channel);
                    subSelector.wakeup();
                    System.out.println("handle Thread:"+Thread.currentThread().getName() + " mainReactor accept remoteAddr:"+ channel.getRemoteAddress().toString()+" and register over!");
                } catch (ClosedChannelException e) {
                    e.printStackTrace();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
        }
    
    }

[https_blog.csdn.net_skh2015java_article_details_106721630]: https://blog.csdn.net/skh2015java/article/details/106721630