0 core classes
NettyWebServer
HttpServer
HttpHandler
TcpServer
1 create a WebServer
According to the article before the Spring Boot startup source code analysis shows that when performing AbstractApplicationContext – > onRefresh () method, if introduced webflux related dependence, will create the WebServer.
/** ReactiveWebServerApplicationContext **/
private void createWebServer(a) {
WebServerManager serverManager = this.serverManager;
if (serverManager == null) {
String webServerFactoryBeanName = getWebServerFactoryBeanName(); / / 1.1
ReactiveWebServerFactory webServerFactory = getWebServerFactory(webServerFactoryBeanName); / / 1.1
boolean lazyInit = getBeanFactory().getBeanDefinition(webServerFactoryBeanName).isLazyInit();
this.serverManager = new WebServerManager(this, webServerFactory, this::getHttpHandler, lazyInit); / / 1.2
getBeanFactory().registerSingleton("webServerGracefulShutdown".new WebServerGracefulShutdownLifecycle(this.serverManager));
getBeanFactory().registerSingleton("webServerStartStop".new WebServerStartStopLifecycle(this.serverManager)); / / 1.3
}
initPropertySources();
}
Copy the code1.1. Select the ReactiveWebServerFactory based on the Web container type. For example, I this is the default Netty, then webServerFactory is NettyReactiveWebServerFactory type.
Create WebServerManager object to manage Server and HttpHandler. The ReactiveWebServerFactory and HttpHandler parameters are obtained from bean container.
1.3, registered WebServerStartStopLifecycle bean, is responsible for the Server to start and stop.
Q: When does the Server start?
A: SpringApplication – > refresh (ConfigurableApplicationContext applicationContext) method, Call AbstractApplicationContext – > finishRefresh (), called after DefaultLifecycleProcessor – > onRefresh (), Called after DefaultLifecycleProcessor – > doStart (Map < String,? Extends Lifecycle> lifecycleBeans, String beanName, Boolean autoStartupOnly) Eventually call in 1.3 WebServerStartStopLifecycle – > start () method, start the service.
So let’s look at the WebServerManager class,
class WebServerManager {
private final ReactiveWebServerApplicationContext applicationContext;
private final DelayedInitializationHttpHandler handler;
private final WebServer webServer;
WebServerManager(ReactiveWebServerApplicationContext applicationContext, ReactiveWebServerFactory factory,
Supplier<HttpHandler> handlerSupplier, boolean lazyInit) {
this.applicationContext = applicationContext;
Assert.notNull(factory, "Factory must not be null");
this.handler = new DelayedInitializationHttpHandler(handlerSupplier, lazyInit);
this.webServer = factory.getWebServer(this.handler); / / 1.4
}
void start(a) {
this.handler.initializeHandler();
this.webServer.start(); / / start the webServer
this.applicationContext
.publishEvent(new ReactiveWebServerInitializedEvent(this.webServer, this.applicationContext)); }... }Copy the codeCreate webServer (NettyWebServer);
/** NettyReactiveWebServerFactory **/
public WebServer getWebServer(HttpHandler httpHandler) {
HttpServer httpServer = createHttpServer(); / / 1.5
ReactorHttpHandlerAdapter handlerAdapter = new ReactorHttpHandlerAdapter(httpHandler);
NettyWebServer webServer = new NettyWebServer(httpServer, handlerAdapter, this.lifecycleTimeout, getShutdown());
webServer.setRouteProviders(this.routeProviders);
return webServer;
}
private HttpServer createHttpServer(a) {
HttpServer server = HttpServer.create(); // 1.6 returns HttpServerBind
if (this.resourceFactory ! =null) { // 1.7 ReactorResourceFactory is a bean
LoopResources resources = this.resourceFactory.getLoopResources();
Assert.notNull(resources, "No LoopResources: is ReactorResourceFactory not initialized yet?");
server = server
.tcpConfiguration((tcpServer) -> tcpServer.runOn(resources).bindAddress(this::getListenAddress)); // 1.8 Returns the HttpServerTcpConfig object
}
else {
server = server.tcpConfiguration((tcpServer) -> tcpServer.bindAddress(this::getListenAddress));
}
if(getSsl() ! =null && getSsl().isEnabled()) {
SslServerCustomizer sslServerCustomizer = new SslServerCustomizer(getSsl(), getHttp2(),
getSslStoreProvider());
server = sslServerCustomizer.apply(server);
}
if(getCompression() ! =null && getCompression().getEnabled()) {
CompressionCustomizer compressionCustomizer = new CompressionCustomizer(getCompression());
server = compressionCustomizer.apply(server);
}
server = server.protocol(listProtocols()).forwarded(this.useForwardHeaders); //1.9 returns the new HttpServerTcpConfig object
return applyCustomizers(server); //1.10 This returns the new HttpServerTcpConfig object
}
Copy the code1.5, create HttpServer, bottom layer is TcpServer, after analysis.
1.7, here early created NettyReactiveWebServerFactory bean. LoopResources manages threads and ConnectionProvider manages connections.
1.8. Bond ports. 1.9. Set the protocol type (Http1.1 or Http2). Note that the incoming arguments are functions that need to wait until they are called.
1.10. Set the length of HTTP request headers.
final class HttpServerTcpConfig extends HttpServerOperator {
final Function<? super TcpServer, ? extends TcpServer> tcpServerMapper;
HttpServerTcpConfig(HttpServer server,
Function<? super TcpServer, ? extends TcpServer> tcpServerMapper) {
super(server);
this.tcpServerMapper = Objects.requireNonNull(tcpServerMapper, "tcpServerMapper");
}
@Override
protected TcpServer tcpConfiguration(a) {
return Objects.requireNonNull(tcpServerMapper.apply(source.tcpConfiguration()),
"tcpServerMapper"); }}abstract class HttpServerOperator extends HttpServer {
final HttpServer source;
HttpServerOperator(HttpServer source) {
this.source = Objects.requireNonNull(source, "source"); }}Copy the code1.8, 1.9, 1.10 return a large number of new HttpServerTcpConfig objects, where 1.5 should return HttpServerTcpConfig. And its source is also of HttpServerTcpConfig type. After multiple layers are nested, the lowest source is of HttpServerBind type.
Continue to see NettyWebServer:
/** NettyWebServer **/
public NettyWebServer(HttpServer httpServer, ReactorHttpHandlerAdapter handlerAdapter, Duration lifecycleTimeout, Shutdown shutdown) {
Assert.notNull(httpServer, "HttpServer must not be null");
Assert.notNull(handlerAdapter, "HandlerAdapter must not be null");
this.lifecycleTimeout = lifecycleTimeout;
this.handler = handlerAdapter;
this.httpServer = httpServer.channelGroup(new DefaultChannelGroup(new DefaultEventExecutor())); / / 1.11
this.gracefulShutdown = (shutdown == Shutdown.GRACEFUL) ? new GracefulShutdown(() -> this.disposableServer) : null;
}
Copy the code1.11, Return to the new HttpServerTcpConfig again where thread pool EventExecutor is created.
2 start the WebServer
/** NettyWebServer **/
private DisposableServer startHttpServer(a) {
HttpServer server = this.httpServer; // HttpServerTcpConfig object in 1.11
if (this.routeProviders.isEmpty()) {
server = server.handle(this.handler); // 2.1 Returns HttpServerHandler extends HttpServerOperator
}
else {
server = server.route(this::applyRouteProviders);
}
if (this.lifecycleTimeout ! =null) {
return server.bindNow(this.lifecycleTimeout);
}
return server.bindNow(); / / 2.2
}
/** HttpServer **/
public final HttpServer handle(BiFunction<? super HttpServerRequest, ? super
HttpServerResponse, ? extends Publisher<Void>> handler) {
return new HttpServerHandle(this, handler);
}
/** HttpServerHandler **/
HttpServerHandle(HttpServer server,
BiFunction<? super HttpServerRequest, ? super
HttpServerResponse, ? extends Publisher<Void>> handler) {
super(server);
this.handler = Objects.requireNonNull(handler, "handler");
}
/** HttpServerOperator **/
HttpServerOperator(HttpServer source) {
this.source = Objects.requireNonNull(source, "source"); // HttpServerTcpConfig object in 1.11
}
Copy the code2.1. Bind handler to HttpServerHandle.
2.2. Bind ports and listen. 2. Bind () and block(timeout) are called.
/** HttpServer **/
public final DisposableServer bindNow(Duration timeout) {
Objects.requireNonNull(timeout, "timeout");
try {
return Objects.requireNonNull(bind().block(timeout), "aborted");
}
catch (IllegalStateException e) {
if (e.getMessage().contains("blocking read")) {
throw new IllegalStateException("HttpServer couldn't be started within "
+ timeout.toMillis() + "ms");
}
throwe; }}Copy the codebind()
Look at the bind() method first:
/** HttpServer **/
public final Mono<? extends DisposableServer> bind() {
return bind(tcpConfiguration());
}
protected TcpServer tcpConfiguration(a) {
return DEFAULT_TCP_SERVER; // It was created in Step 1.6
}
static final TcpServer DEFAULT_TCP_SERVER = TcpServer.create();
/** TcpServer **/
public static TcpServer create(a) {
return TcpServerBind.INSTANCE; / / 2.3
}
static final TcpServerBind INSTANCE = new TcpServerBind();
TcpServerBind() {
this.serverBootstrap = createServerBootstrap();
BootstrapHandlers.channelOperationFactory(this.serverBootstrap, TcpUtils.TCP_OPS);
}
ServerBootstrap createServerBootstrap(a) { // 2.4 Invalid ServerBootstrap
return new ServerBootstrap()
.option(ChannelOption.SO_REUSEADDR, true)
.childOption(ChannelOption.AUTO_READ, false)
.childOption(ChannelOption.TCP_NODELAY, true)
.localAddress(new InetSocketAddress(DEFAULT_PORT));
}
Copy the codeThe TcpServerBind.INSTANCE object was already created in advance when the HttpServerBind object was created in Step 1.6.
According to OSI’s seven-layer network model, ports are defined at the transport layer, so the bonding of ports should be implemented at the TCP layer. So the first thing you do here is create a TcpServerBind object that takes care of this. In this class, there is the ServerBootstrap property, which can be used to concatenate EventLoopGroups, SocketChannels, and Handlers in Netty. A typical Netty startup code looks like this:
EventLoopGroup boss = new NioEventLoopGroup();
EventLoopGroup worker = new NioEventLoopGroup();
try {
ServerBootstrap b = new ServerBootstrap();
b.group(boss, worker) // EventLoopGroup
.channel(NioServerSocketChannel.class) // SocketChannel
.localAddress(new InetSocketAddress(port)) // ip+port
.childHandler(new ChannelInitializer<SocketChannel>() { // ChannelHandler
@Override
protected void initChannel(SocketChannel socketChannel) throws Exception {... }}); ChannelFuture future = b.bind().sync(); future.channel().closeFuture().sync(); }finally {
group.shutdownGracefully().sync();
worker.shutdownGracefully().sync();
}
Copy the codeAs you can see, the ServerBootstrap property in the created tcpServerBind.instance is missing a lot of things: EventLoopGroup, SocketChannel, and Handler.
The bind method is of type HttpServerHandler, and its source is of type HttpServerTcpConfig. Its final source is of type HttpServerBind.
/** HttpServer **/
public final Mono<? extends DisposableServer> bind() {
return bind(tcpConfiguration()); What type of TcpServer is returned?
}
/** HttpServerHandler **/
protected TcpServer tcpConfiguration(a) {
return source.tcpConfiguration().bootstrap(this); // 2.6 Return TcpServerBootstrap
}
/** HttpServerTcpConfig **/
protected TcpServer tcpConfiguration(a) {
return Objects.requireNonNull(tcpServerMapper.apply(source.tcpConfiguration()),
"tcpServerMapper"); // 2.7 Return TcpServerBootstrap or TcpServerChannelGroup(Step 1.11)
}
/** HttpServerBind **/
protected TcpServer tcpConfiguration(a) {
return tcpServer;
}
Copy the codeTo obtain the TcpServer in Step 2.5, you need to execute HttpServerHandler -> tcpConfiguration() and HttpServerTcpConfig -> tcpConfiguration() several times. In Step 2.7, the previous functions (steps 1.8, 1.9, 1.11, etc.) are called.
Conclusion: Step 2.5 returns the TcpServerBootstrap type, and the source is also the TcpServerBootstrap type (the TcpServerChannelGroup type is returned in Step 1.11, which is the main analysis returned in Step 1.8). After multiple nesting, the final source is TcpServerBind.
Read on:
/** HttpServerBind **/
public Mono<? extends DisposableServer> bind(TcpServer delegate) {
return delegate.bootstrap(this) / / return TcpServerBootstrap
.bind()
.map(CLEANUP_GLOBAL_RESOURCE);
}
/** TcpServer **/
public final TcpServer bootstrap(Function<? super ServerBootstrap, ? extends ServerBootstrap> bootstrapMapper) {
return new TcpServerBootstrap(this, bootstrapMapper);
}
public final Mono<? extends DisposableServer> bind() {
ServerBootstrap b;
try{
b = configure(); / / 2.8
}
catch (Throwable t){
Exceptions.throwIfJvmFatal(t);
return Mono.error(t);
}
return bind(b); / / 2.12
}
/** TcpServerBootstrap **/
public ServerBootstrap configure(a) {
return Objects.requireNonNull(bootstrapMapper.apply(source.configure()), "bootstrapMapper"); / / 2.9
}
/** TcpServerChannelGroup **/
public ServerBootstrap configure(a) { / / 2.10
ServerBootstrap b = source.configure();
b.attr(CHANNEL_GROUP, channelGroup);
ConnectionObserver observer = BootstrapHandlers.childConnectionObserver(b);
BootstrapHandlers.childConnectionObserver(b, observer.then(this));
return b;
}
/** TcpServerBind **/
public ServerBootstrap configure(a) { / / 2.11
return this.serverBootstrap.clone();
}
Copy the code2.8. Call the configure() method in 2.9-2.11. This method configures ServerBootstrap with childOptions. The ServerBootstrap group and Channel properties are filled in here, but handler is missing.
/** NettyReactiveWebServerFactory **/
server = server
.tcpConfiguration((tcpServer) -> tcpServer.runOn(resources).bindAddress(this::getListenAddress));
/** TcpServer **/
public final TcpServer runOn(LoopResources channelResources) {
return runOn(channelResources, LoopResources.DEFAULT_NATIVE);
}
public final TcpServer runOn(LoopResources channelResources, boolean preferNative) {
return new TcpServerRunOn(this, channelResources, preferNative);
}
/** TcpServerRunOn **/
TcpServerRunOn(TcpServer server, LoopResources loopResources, boolean preferNative) {
super(server);
this.loopResources = Objects.requireNonNull(loopResources, "loopResources");
this.preferNative = preferNative;
}
public ServerBootstrap configure(a) {
ServerBootstrap b = source.configure();
configure(b, preferNative, loopResources);
return b;
}
static void configure(ServerBootstrap b,
boolean preferNative,
LoopResources resources) {
EventLoopGroup selectorGroup = resources.onServerSelect(preferNative);
EventLoopGroup elg = resources.onServer(preferNative);
b.group(selectorGroup, elg) // Familiar with Netty code
.channel(resources.onServerChannel(elg));
}
Copy the codeBootstrapmapper.apply (source.configure()), bootstrapmapper.apply (source.configure()), bootstrapmapper.apply (source.configure()), bootstrapmapper.apply (source.configure())), bootstrapmapper.configure ())
/** HttpServerBind **/
public ServerBootstrap apply(ServerBootstrap b) {
HttpServerConfiguration conf = HttpServerConfiguration.getAndClean(b); / / 2.12
SslProvider ssl = SslProvider.findSslSupport(b); / / 2.13
if(ssl ! =null && ssl.getDefaultConfigurationType() == null) {
if ((conf.protocols & HttpServerConfiguration.h2) == HttpServerConfiguration.h2) {
ssl = SslProvider.updateDefaultConfiguration(ssl,
SslProvider.DefaultConfigurationType.H2);
SslProvider.setBootstrap(b, ssl);
}
else{ ssl = SslProvider.updateDefaultConfiguration(ssl, SslProvider.DefaultConfigurationType.TCP); SslProvider.setBootstrap(b, ssl); }}if (b.config()
.group() == null) { / / 2.14
LoopResources loops = HttpResources.get();
EventLoopGroup selector = loops.onServerSelect(LoopResources.DEFAULT_NATIVE);
EventLoopGroup elg = loops.onServer(LoopResources.DEFAULT_NATIVE);
b.group(selector, elg)
.channel(loops.onServerChannel(elg));
}
//remove any OPS since we will initialize below
BootstrapHandlers.channelOperationFactory(b);
if(ssl ! =null) {
if ((conf.protocols & HttpServerConfiguration.h2c) == HttpServerConfiguration.h2c) { / / 2.15
throw new IllegalArgumentException("Configured H2 Clear-Text protocol " +
"with TLS. Use the non clear-text h2 protocol via " +
"HttpServer#protocol or disable TLS" +
" via HttpServer#tcpConfiguration(tcp -> tcp.noSSL())");
}
if ((conf.protocols & HttpServerConfiguration.h11orH2) == HttpServerConfiguration.h11orH2) {
return BootstrapHandlers.updateConfiguration(b,
NettyPipeline.HttpInitializer,
new Http1OrH2Initializer(conf.decoder.maxInitialLineLength(),
conf.decoder.maxHeaderSize(),
conf.decoder.maxChunkSize(),
conf.decoder.validateHeaders(),
conf.decoder.initialBufferSize(),
conf.minCompressionSize,
compressPredicate(conf.compressPredicate, conf.minCompressionSize),
conf.forwarded,
conf.cookieEncoder,
conf.cookieDecoder,
conf.uriTagValue));
}
if ((conf.protocols & HttpServerConfiguration.h11) == HttpServerConfiguration.h11) {
return BootstrapHandlers.updateConfiguration(b,
NettyPipeline.HttpInitializer,
new Http1Initializer(conf.decoder.maxInitialLineLength(),
conf.decoder.maxHeaderSize(),
conf.decoder.maxChunkSize(),
conf.decoder.validateHeaders(),
conf.decoder.initialBufferSize(),
conf.minCompressionSize,
compressPredicate(conf.compressPredicate, conf.minCompressionSize),
conf.forwarded,
conf.cookieEncoder,
conf.cookieDecoder,
conf.uriTagValue));
}
if ((conf.protocols & HttpServerConfiguration.h2) == HttpServerConfiguration.h2) {
return BootstrapHandlers.updateConfiguration(b,
NettyPipeline.HttpInitializer,
newH2Initializer( conf.decoder.validateHeaders(), conf.minCompressionSize, compressPredicate(conf.compressPredicate, conf.minCompressionSize), conf.forwarded, conf.cookieEncoder, conf.cookieDecoder)); }}else {
if ((conf.protocols & HttpServerConfiguration.h2) == HttpServerConfiguration.h2) {
throw new IllegalArgumentException(
"Configured H2 protocol without TLS. Use" +
" a clear-text h2 protocol via HttpServer#protocol or configure TLS" +
" via HttpServer#secure");
}
if ((conf.protocols & HttpServerConfiguration.h11orH2c) == HttpServerConfiguration.h11orH2c) {
return BootstrapHandlers.updateConfiguration(b,
NettyPipeline.HttpInitializer,
new Http1OrH2CleartextInitializer(conf.decoder.maxInitialLineLength(),
conf.decoder.maxHeaderSize(),
conf.decoder.maxChunkSize(),
conf.decoder.validateHeaders(),
conf.decoder.initialBufferSize(),
conf.minCompressionSize,
compressPredicate(conf.compressPredicate, conf.minCompressionSize),
conf.forwarded,
conf.cookieEncoder,
conf.cookieDecoder,
conf.uriTagValue,
conf.decoder.h2cMaxContentLength));
}
if ((conf.protocols & HttpServerConfiguration.h11) == HttpServerConfiguration.h11) { / / 2.16
return BootstrapHandlers.updateConfiguration(b,
NettyPipeline.HttpInitializer,
new Http1Initializer(conf.decoder.maxInitialLineLength(),
conf.decoder.maxHeaderSize(),
conf.decoder.maxChunkSize(),
conf.decoder.validateHeaders(),
conf.decoder.initialBufferSize(),
conf.minCompressionSize,
compressPredicate(conf.compressPredicate, conf.minCompressionSize),
conf.forwarded,
conf.cookieEncoder,
conf.cookieDecoder,
conf.uriTagValue));
}
if ((conf.protocols & HttpServerConfiguration.h2c) == HttpServerConfiguration.h2c) {
return BootstrapHandlers.updateConfiguration(b,
NettyPipeline.HttpInitializer,
newH2CleartextInitializer( conf.decoder.validateHeaders(), conf.minCompressionSize, compressPredicate(conf.compressPredicate, conf.minCompressionSize), conf.forwarded, conf.cookieEncoder, conf.cookieDecoder)); }}throw new IllegalArgumentException("An unknown HttpServer#protocol " +
"configuration has been provided: "+String.format("0x%x", conf
.protocols));
}
Copy the code2.12 ServerBootstrapConfig can be used to obtain serverBootstrap-related information, such as childGroup, childHanlder, childOptions, childAttrs. The attrs -> httpServerConf attribute in the original ServerBootstrap has been cleared.
2.13 Check whether SSL is supported. If SSL is supported, add related handlers.
2.14. It has been set before;
2.15 If SSL is set, then H2C protocol is not supported. There are three main HTTP protocols: H11 (http1.1), H2 (http2), AND H2C (plaintext version of http2).
2.16. Select an appropriate childHandler based on the protocol type. For example, set Http1Initializer for http1.1 plaintext.
At this point, a complete ServerBootstrap is generated. Now you can bind().
/** TcpServerBind **/
public Mono<? extends DisposableServer> bind(ServerBootstrap b) {
SslProvider ssl = SslProvider.findSslSupport(b);
if(ssl ! =null && ssl.getDefaultConfigurationType() == null) {
ssl = SslProvider.updateDefaultConfiguration(ssl, SslProvider.DefaultConfigurationType.TCP);
SslProvider.setBootstrap(b, ssl);
}
if (b.config()
.group() == null) {
TcpServerRunOn.configure(b, LoopResources.DEFAULT_NATIVE, TcpResources.get());
}
return Mono.create(sink -> { / / 2.17
ServerBootstrap bootstrap = b.clone();
ConnectionObserver obs = BootstrapHandlers.connectionObserver(bootstrap);
ConnectionObserver childObs =
BootstrapHandlers.childConnectionObserver(bootstrap);
ChannelOperations.OnSetup ops =
BootstrapHandlers.channelOperationFactory(bootstrap);
convertLazyLocalAddress(bootstrap);
BootstrapHandlers.finalizeHandler(bootstrap, ops, new ChildObserver(childObs));
ChannelFuture f = bootstrap.bind(); // The port is formally bound
DisposableBind disposableServer = new DisposableBind(sink, f, obs, bootstrap);
f.addListener(disposableServer);
sink.onCancel(disposableServer);
});
}
Copy the code2.17. Note that the Mono type is returned and a subscription is required to trigger the formal binding port operation.
block()
/** Mono **/
public T block(Duration timeout) {
BlockingMonoSubscriber<T> subscriber = new BlockingMonoSubscriber<>();
subscribe((Subscriber<T>) subscriber);
return subscriber.blockingGet(timeout.toMillis(), TimeUnit.MILLISECONDS);
}
abstract class BlockingSingleSubscriber<T> extends CountDownLatch
implements InnerConsumer<T>, Disposable {
T value;
public final void onSubscribe(Subscription s) {
this.s = s;
if(! cancelled) { s.request(Long.MAX_VALUE);// Initiate a request, triggering 2.17 Execution}}Copy the code3 source code design ideas
-
layered
The TcpServer performs functions at the TCP layer, such as binding ports and setting parameters TCP_NODELAY at the TCP layer. HttpServer is responsible for HTTP protocol functions, such as codec.
-
Points module
For example, TcpServerBind is responsible for port binding and TcpServerRunOn is responsible for thread pool binding.
Likewise, HttpServer has a similar setup.
-
Chain calls
The configure() or tcpConfiguration() methods of TcpServer and HttpServer are chain-called with nested wraps.
-
Method input arguments are functions, which are triggered by subscriptions.