io.netty.channel.kqueue.KQueueServerSocketChannel#io.netty.channel.socket.ServerSocketChannel源码实例Demo

/**
 * 启动端口绑定
 * @param local
 * @return
 */
protected final boolean bind(InetSocketAddress local)
{
	boolean isBind=false;
	try {
		log.debug(getName()+"端口绑定中……"+local.toString());
		ChannelFuture cf=doBind(local);
		isBind=cf.channel()!=null && cf.channel().isActive();
		if(isBind)
		{
			log.debug(getName()+"端口绑定成功!"+cf.channel());
			serverCahnel=(ServerSocketChannel) cf.channel();
		}else
		{
			log.debug(getName()+"端口绑定失败!"+cf.channel());
		}
	} catch (Exception e) {
		log.error(e.getMessage(),e);
		throw e;
	}
	return isBind;
}
 

@Override
@SuppressWarnings("unchecked")
public <CHANNEL extends Channel> CHANNEL getChannel(Class<CHANNEL> channelClass) {
	if (channelClass.equals(SocketChannel.class)) {
		return (CHANNEL) new EpollSocketChannel();
	}
	if (channelClass.equals(ServerSocketChannel.class)) {
		return (CHANNEL) new EpollServerSocketChannel();
	}
	if (channelClass.equals(DatagramChannel.class)) {
		return (CHANNEL) new EpollDatagramChannel();
	}
	if (channelClass.equals(DomainSocketChannel.class)) {
		return (CHANNEL) new EpollDomainSocketChannel();
	}
	if (channelClass.equals(ServerDomainSocketChannel.class)) {
		return (CHANNEL) new EpollServerDomainSocketChannel();
	}
	throw new IllegalArgumentException("Unsupported channel type: " + channelClass.getSimpleName());
}
 

@Override
@SuppressWarnings("unchecked")
public <CHANNEL extends Channel> CHANNEL getChannel(Class<CHANNEL> channelClass) {
	if (channelClass.equals(SocketChannel.class)) {
		return (CHANNEL) new KQueueSocketChannel();
	}
	if (channelClass.equals(ServerSocketChannel.class)) {
		return (CHANNEL) new KQueueServerSocketChannel();
	}
	if (channelClass.equals(DatagramChannel.class)) {
		return (CHANNEL) new KQueueDatagramChannel();
	}
	if (channelClass.equals(DomainSocketChannel.class)) {
		return (CHANNEL) new KQueueDomainSocketChannel();
	}
	if (channelClass.equals(ServerDomainSocketChannel.class)) {
		return (CHANNEL) new KQueueServerDomainSocketChannel();
	}
	throw new IllegalArgumentException("Unsupported channel type: " + channelClass.getSimpleName());
}
 

protected void startServer() throws InterruptedException {

      ServerBootstrap b = new ServerBootstrap();
      b.group(bossGroup, workerGroup)
       .channel(NioServerSocketChannel.class)
       .handler(new LoggingHandler(LogLevel.INFO))
       .childHandler(new ChannelInitializer<SocketChannel>() {
           @Override
           public void initChannel(SocketChannel ch) throws Exception {
               ChannelPipeline p = ch.pipeline();
               p.addLast(new LoggingHandler(LogLevel.DEBUG));
               p.addLast(new NettySimpleMessageHandler());
               p.addLast(new NettyMasterHandler(DefaultRedisKeeperServer.this, new CommandHandlerManager(), keeperConfig.getTrafficReportIntervalMillis()));
           }
       });
      serverSocketChannel = (ServerSocketChannel) b.bind(currentKeeperMeta.getPort()).sync().channel();
  }
 

public static Class<? extends ServerSocketChannel> getServerSocketChannel(){
  if(SUPPORTS_EPOLL){
    return EpollServerSocketChannel.class;
  }else{
    return NioServerSocketChannel.class;
  }
}
 

public NettyConfiguration(final Class<? extends ServerSocketChannel> serverSocketChannelClass,
                          final Class<? extends SocketChannel> clientSocketChannelClass,
                          final EventLoopGroup parentEventLoopGroup,
                          final EventLoopGroup childEventLoopGroup) {

    checkNotNull(serverSocketChannelClass, "Server Socket Channel Class must not be null");
    checkNotNull(clientSocketChannelClass, "Client Socket Channel Class must not be null");
    checkNotNull(parentEventLoopGroup, "Parent Event Loop Group must not be null");
    checkNotNull(childEventLoopGroup, "Child Event Loop Group must not be null");

    this.serverSocketChannelClass = serverSocketChannelClass;
    this.clientSocketChannelClass = clientSocketChannelClass;
    this.parentEventLoopGroup = parentEventLoopGroup;
    this.childEventLoopGroup = childEventLoopGroup;
}
 

private ServerSocketChannel startServer() {
    EventLoopAwareNettyIoExecutor eventLoopAwareNettyIoExecutor =
            toEventLoopAwareNettyIoExecutor(S_CTX.ioExecutor());
    EventLoop loop = eventLoopAwareNettyIoExecutor.eventLoopGroup().next();

    ServerBootstrap bs = new ServerBootstrap();
    bs.group(loop);
    bs.channel(serverChannel(loop, InetSocketAddress.class));
    bs.childHandler(new ChannelInitializer() {
        @Override
        protected void initChannel(final Channel ch) {
            sChannel = (SocketChannel) ch;
            ch.pipeline().addLast(new ChannelInboundHandlerAdapter() {
                @Override
                public void userEventTriggered(final ChannelHandlerContext ctx, final Object evt) {
                    LOGGER.debug("Server Evt: {}", evt.getClass().getSimpleName());
                    if (evt == ChannelInputShutdownEvent.INSTANCE) {
                        serverInputShutdownLatch.countDown();
                    } else if (evt == ChannelInputShutdownReadComplete.INSTANCE) {
                        serverInputShutdownReadCompleteLatch.countDown();
                    } else if (evt == ChannelOutputShutdownEvent.INSTANCE) {
                        serverOutputShutdownLatch.countDown();
                    }
                    release(evt);
                }
            });
            ch.eventLoop().execute(connectedLatch::countDown);
        }
    });

    bs.childOption(AUTO_READ, true);
    bs.childOption(ALLOW_HALF_CLOSURE, true);
    bs.childOption(AUTO_CLOSE, false);

    return (ServerSocketChannel) bs.bind(localAddress(0))
            .syncUninterruptibly().channel();
}
 

public PrematureClosureBeforeResponsePayloadBodyTest() {
    EventLoopAwareNettyIoExecutor eventLoopAwareNettyIoExecutor =
            toEventLoopAwareNettyIoExecutor(globalExecutionContext().ioExecutor());
    EventLoop loop = eventLoopAwareNettyIoExecutor.eventLoopGroup().next();

    ServerBootstrap bs = new ServerBootstrap();
    bs.group(loop);
    bs.channel(serverChannel(loop, InetSocketAddress.class));
    bs.childHandler(new ChannelInitializer() {
        @Override
        protected void initChannel(Channel ch) {
            ch.pipeline().addLast(new HttpRequestDecoder());
            ch.pipeline().addLast(new HttpObjectAggregator(MAX_VALUE));
            ch.pipeline().addLast(new ChannelInboundHandlerAdapter() {
                @Override
                public void channelRead(final ChannelHandlerContext ctx, final Object msg) {
                    if (msg instanceof FullHttpRequest) {
                        ctx.writeAndFlush(ByteBufUtil.writeAscii(ctx.alloc(), encodedResponse.get()))
                                .addListener(ChannelFutureListener.CLOSE);
                    }
                    ReferenceCountUtil.release(msg);
                }
            });
        }
    });
    bs.childOption(AUTO_READ, true);
    bs.childOption(ALLOW_HALF_CLOSURE, true);
    bs.childOption(AUTO_CLOSE, false);
    server = (ServerSocketChannel) bs.bind(localAddress(0))
            .syncUninterruptibly().channel();

    client = HttpClients.forSingleAddress(HostAndPort.of(server.localAddress()))
            .protocols(h1()
                    .specExceptions(new H1SpecExceptions.Builder().allowPrematureClosureBeforePayloadBody().build())
                    .build())
            .buildBlocking();
}
 

@Override
public final void run() {
    try {
        if(running){
            return;
        }
        init();
        ChannelFuture channelFuture = bootstrap.bind(serverAddress).addListener((ChannelFutureListener) this::startAfter);
        this.serverChannel = (ServerSocketChannel) channelFuture.channel();
        this.running = true;
    } catch (Throwable throwable) {
        logger.error("server run fail. cause={}",throwable.toString(),throwable);
    }
}
 

private static Class<? extends ServerSocketChannel> serverSocketChannelType(
        final OptimalTransport optimalTransport,
        final Logger logger) {

  switch (optimalTransport) {
  case Epoll:
    logger.debug("HttpAgent using EpollServerSocketChannel");
    return EpollServerSocketChannel.class;
  case NIO:
  default:
    logger.debug("HttpAgent using NioServerSocketChannel");
    return NioServerSocketChannel.class;
  }
}
 

<T extends ServerSocketChannel> RpcServer(EventLoopGroup eventLoopGroup, EventExecutorGroup eventExecutor, Class<T> channel, SocketAddress address) {
    this.address = address;
    this.allChannels = new DefaultChannelGroup(eventLoopGroup.next());
    this.handler = new ServerHandler(allChannels);
    this.bootstrap = new ServerBootstrap();
    bootstrap.channel(channel);
    bootstrap.childHandler(new ServerInitializer(eventExecutor, handler));
    bootstrap.group(eventLoopGroup);
    bootstrap.option(ChannelOption.TCP_NODELAY, true);
}
 

public void init() throws Exception {
    SelfSignedCertificate ssc = new SelfSignedCertificate();
    sslCtx = SslContextBuilder.forServer(ssc.certificate(), ssc.privateKey()).build();

    bootstrap = new ServerBootstrap()
        .channel(NioServerSocketChannel.class)
        .handler(new LoggingHandler(LogLevel.DEBUG))
        .group(group)
        .childHandler(this);

    serverSock = (ServerSocketChannel) bootstrap.bind(0).sync().channel();
}
 

public void init() throws InterruptedException {
    bootstrap = new ServerBootstrap()
            .channel(NioServerSocketChannel.class)
            .group(new NioEventLoopGroup())
            .childHandler(this)
            .localAddress(0)
            .childOption(ChannelOption.SO_KEEPALIVE, true);

    channel = ((ServerSocketChannel) bootstrap.bind().await().channel());
}
 

public void init() throws InterruptedException {
    bootstrap = new ServerBootstrap()
            .channel(NioServerSocketChannel.class)
            .group(new NioEventLoopGroup())
            .childHandler(this)
            .localAddress(0)
            .childOption(ChannelOption.SO_KEEPALIVE, true);

    channel = ((ServerSocketChannel) bootstrap.bind().await().channel());
}
 

public void init() throws Exception {
    SelfSignedCertificate ssc = new SelfSignedCertificate();
    sslCtx = SslContextBuilder.forServer(ssc.certificate(), ssc.privateKey()).build();

    bootstrap = new ServerBootstrap()
        .channel(NioServerSocketChannel.class)
        .handler(new LoggingHandler(LogLevel.DEBUG))
        .group(group)
        .childHandler(this);

    serverSock = (ServerSocketChannel) bootstrap.bind(0).sync().channel();
}
 

void init() throws Exception {
    SelfSignedCertificate ssc = new SelfSignedCertificate();
    bootstrap = new ServerBootstrap()
            .channel(NioServerSocketChannel.class)
            .group(group)
            .childHandler(this);

    serverSock = (ServerSocketChannel) bootstrap.bind(0).sync().channel();
}
 

void init() throws Exception {
    SelfSignedCertificate ssc = new SelfSignedCertificate();
    sslCtx = SslContextBuilder.forServer(ssc.certificate(), ssc.privateKey()).build();

    bootstrap = new ServerBootstrap()
        .channel(NioServerSocketChannel.class)
        .group(group)
        .childHandler(this);

    serverSock = (ServerSocketChannel) bootstrap.bind(0).sync().channel();
}
 

void init() throws Exception {
    SelfSignedCertificate ssc = new SelfSignedCertificate();
    sslCtx = SslContextBuilder.forServer(ssc.certificate(), ssc.privateKey()).build();

    bootstrap = new ServerBootstrap()
        .channel(NioServerSocketChannel.class)
        .group(group)
        .childHandler(this);

    serverSock = (ServerSocketChannel) bootstrap.bind(0).sync().channel();
}
 

void init() throws Exception {
    SelfSignedCertificate ssc = new SelfSignedCertificate();
    sslCtx = SslContextBuilder.forServer(ssc.certificate(), ssc.privateKey()).build();

    bootstrap = new ServerBootstrap()
        .channel(NioServerSocketChannel.class)
        .group(group)
        .childHandler(this);

    serverSock = (ServerSocketChannel) bootstrap.bind(0).sync().channel();
}
 

public void init() throws InterruptedException {
    bootstrap = new ServerBootstrap()
            .channel(NioServerSocketChannel.class)
            .group(new NioEventLoopGroup())
            .childHandler(this)
            .childOption(ChannelOption.SO_KEEPALIVE, true);

    serverSock = (ServerSocketChannel) bootstrap.bind(0).sync().channel();
}
 

public static Class<? extends ServerSocketChannel> getServerSocketChannel(){
  if(SUPPORTS_EPOLL){
    return EpollServerSocketChannel.class;
  }else{
    return NioServerSocketChannel.class;
  }
}
 

@Override
@SuppressWarnings("unchecked")
public <CHANNEL extends Channel> Class<? extends CHANNEL> getChannelClass(Class<CHANNEL> channelClass) {
	if (channelClass.equals(SocketChannel.class)) {
		return (Class<? extends CHANNEL>) EpollSocketChannel.class;
	}
	if (channelClass.equals(ServerSocketChannel.class)) {
		return (Class<? extends CHANNEL>) EpollServerSocketChannel.class;
	}
	if (channelClass.equals(DatagramChannel.class)) {
		return (Class<? extends CHANNEL>) EpollDatagramChannel.class;
	}
	throw new IllegalArgumentException("Unsupported channel type: " + channelClass.getSimpleName());
}
 

@Override
@SuppressWarnings("unchecked")
public <CHANNEL extends Channel> Class<? extends CHANNEL> getChannelClass(Class<CHANNEL> channelClass) {
	if (channelClass.equals(SocketChannel.class)) {
		return (Class<? extends CHANNEL>) KQueueSocketChannel.class;
	}
	if (channelClass.equals(ServerSocketChannel.class)) {
		return (Class<? extends CHANNEL>) KQueueServerSocketChannel.class;
	}
	if (channelClass.equals(DatagramChannel.class)) {
		return (Class<? extends CHANNEL>) KQueueDatagramChannel.class;
	}
	throw new IllegalArgumentException("Unsupported channel type: " + channelClass.getSimpleName());
}
 

@Override
@SuppressWarnings("unchecked")
public <CHANNEL extends Channel> CHANNEL getChannel(Class<CHANNEL> channelClass) {
	if (channelClass.equals(SocketChannel.class)) {
		return (CHANNEL) new NioSocketChannel();
	}
	if (channelClass.equals(ServerSocketChannel.class)) {
		return (CHANNEL) new NioServerSocketChannel();
	}
	if (channelClass.equals(DatagramChannel.class)) {
		return (CHANNEL) new NioDatagramChannel();
	}
	throw new IllegalArgumentException("Unsupported channel type: " + channelClass.getSimpleName());
}
 

@Override
@SuppressWarnings("unchecked")
public <CHANNEL extends Channel> Class<? extends CHANNEL> getChannelClass(Class<CHANNEL> channelClass) {
	if (channelClass.equals(SocketChannel.class)) {
		return (Class<? extends CHANNEL>) NioSocketChannel.class;
	}
	if (channelClass.equals(ServerSocketChannel.class)) {
		return (Class<? extends CHANNEL>) NioServerSocketChannel.class;
	}
	if (channelClass.equals(DatagramChannel.class)) {
		return (Class<? extends CHANNEL>) NioDatagramChannel.class;
	}
	throw new IllegalArgumentException("Unsupported channel type: " + channelClass.getSimpleName());
}
 

public static Class<? extends ServerSocketChannel> getServerSocketChannelClass(EventLoopGroup eventLoopGroup) {
    if (eventLoopGroup instanceof EpollEventLoopGroup) {
        return EpollServerSocketChannel.class;
    } else {
        return NioServerSocketChannel.class;
    }
}
 

/**
 * Bootstrap a new server with a {@link ChannelInitializer} and bind it to a port.
 * @param port the port number to bind this server to.
 * @param channelInitializer the {@link ChannelInitializer} for request handling on this server.
 * @param bossGroup the pool of boss threads that this server uses.
 * @param workerGroup the pool of worker threads that this server uses.
 * @throws InterruptedException if binding to the port failed.
 */
private void bindServer(int port, ChannelInitializer<SocketChannel> channelInitializer, EventLoopGroup bossGroup,
    EventLoopGroup workerGroup) throws InterruptedException {
  ServerBootstrap b = new ServerBootstrap();
  Class<? extends ServerSocketChannel> channelClass =
      Epoll.isAvailable() ? EpollServerSocketChannel.class : NioServerSocketChannel.class;
  // Note: ServerSocketChannel option doesn't apply to SocketChannel
  b.group(bossGroup, workerGroup)
      .channel(channelClass)
      .option(ChannelOption.SO_BACKLOG, nettyConfig.nettyServerSoBacklog)
      .handler(new LoggingHandler(LogLevel.DEBUG))
      .childHandler(channelInitializer);
  b.bind(port).sync();
  logger.info("NettyServer now listening on port {}", port);
}
 

@Override
protected boolean isCompatible(EventLoop loop) {
    if (!(loop instanceof XnioEventLoop)) {
        return false;
    }
    ServerSocketChannel parent = parent();
    if (parent != null) {
        // if this channel has a parent we need to ensure that both EventLoopGroups are the same for XNIO
        // to be sure it uses a Thread from the correct Worker.
        if (parent.eventLoop().parent() != loop.parent()) {
            return false;
        }
    }
    return true;
}
 

public Class<? extends ServerSocketChannel> getServerSocketChannelClass() {
    return serverSocketChannelClass;
}
 

@Override
public Class<? extends ServerSocketChannel> getServerSocketChannelClass() {
   return NioServerSocketChannel.class;
}