Reactor模式笔记

{
    // 创建一个serverSocket对象,相当于服务器，并且自己设定端口，最好设置1024以后
    ServerSocket serverSocket = new ServerSocket(8888);

   while (true){
     // 调用accept方法监听访问的Socket
     Socket socket = serverSocket.accept();
      System.out.println("接受到新socket...");

      new Thread(
              new Runnable() {
                @Override
                public void run() {
                  try {
                    // 从socket中读数据
                    InputStream is = socket.getInputStream();
                    InputStreamReader isr = new InputStreamReader(is);
                    BufferedReader bs = new BufferedReader(isr);
                    String str = "";
                    while ((str = bs.readLine()) != null) {
                      System.out.println("我是服务器，客户端说：" + str);
                    }

                    // 关闭输入流
                    socket.shutdownInput();

                    // ---->下面是服务器响应客户端
                    // 获得输出流
                    OutputStream os = socket.getOutputStream();
                    // 写入数据
                    PrintWriter pw = new PrintWriter(os);
                    pw.write("欢迎您：" + new Date().toString());
                    pw.flush();

                    // 关闭输出流资源
                    socket.shutdownOutput();
                    pw.close();
                    os.close();

                    // 关闭输入流资源
                    bs.close();
                    isr.close();
                    is.close();
                  } catch (Exception e) {
                    e.printStackTrace();
                  }
                }
              })
          .start();
   }
  }

{

    ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();

    serverSocketChannel.configureBlocking(false);

    serverSocketChannel.bind(new InetSocketAddress(10022));

    Selector selector = Selector.open();

    serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);
    while (selector.select() > 0) {
      Iterator<SelectionKey> iterator = selector.selectedKeys().iterator();
      while (iterator.hasNext()) {
        SelectionKey next = iterator.next();
        if (next.isAcceptable()) {
          SocketChannel socketChannel = serverSocketChannel.accept();

          socketChannel.configureBlocking(false);

          socketChannel.register(selector, SelectionKey.OP_READ);

        } else if (next.isReadable()) {
          SocketChannel socketChannel = (SocketChannel) next.channel();

          // 分配指定大小缓冲区
          ByteBuffer buffer = ByteBuffer.allocate(1024 * 5);
          int len = 0;
          while ((len = socketChannel.read(buffer)) > 0) {

            buffer.flip();
            System.out.println(new String(buffer.array(), 0, len));
            buffer.clear();
          }
        }
        iterator.remove();
      }
    }
  }

EventLoopGroup bossGroup = new NioEventLoopGroup();
    EventLoopGroup workerGroup = new NioEventLoopGroup();

    try {

      ServerBootstrap b = new ServerBootstrap();
      b.group(bossGroup, workerGroup)
          .channel(NioServerSocketChannel.class)
          .option(ChannelOption.SO_BACKLOG, 1024)
          // 保持连接
          .childOption(ChannelOption.SO_KEEPALIVE, true)
          .childHandler(
              new ChannelInitializer<SocketChannel>() {
                @Override
                protected void initChannel(SocketChannel socketChannel) throws Exception {
                  // TimeClientHandler是自己定义的方法
                  socketChannel.pipeline().addLast(new StringDecoder(CharsetUtil.UTF_8));
                  socketChannel.pipeline().addLast(new StringEncoder(CharsetUtil.UTF_8));
                  socketChannel.pipeline().addLast(new MyServerHandler());
                }
              });
      // 绑定端口
      ChannelFuture f = b.bind(8888).sync();
      // 等待服务端监听端口关闭
      f.channel().closeFuture().sync();

    } catch (Exception e) {

    } finally {
      // 优雅关闭，释放线程池资源
      bossGroup.shutdownGracefully();
      workerGroup.shutdownGracefully();
    }

//public class ServerBootstrap extends AbstractBootstrap<ServerBootstrap, ServerChannel>
@Override
    void init(Channel channel) throws Exception {
        
         //省略

        p.addLast(new ChannelInitializer<Channel>() {
            @Override
            public void initChannel(final Channel ch) throws Exception {
                final ChannelPipeline pipeline = ch.pipeline();
                ChannelHandler handler = config.handler();
                if (handler != null) {
                    pipeline.addLast(handler);
                }

                ch.eventLoop().execute(new Runnable() {
                    @Override
                    public void run() {
                       //acceptor
                       //acceptor
                       //acceptor
                        pipeline.addLast(new ServerBootstrapAcceptor(
                                ch, currentChildGroup, currentChildHandler, currentChildOptions, currentChildAttrs));
                    }
                });
            }
        });
    }