序
本文主要研究一下NettyConnector的start及shutdown
NettyConnector
reactor-netty-0.7.6.RELEASE-sources.jar!/reactor/ipc/netty/NettyConnector.java
/** * A Netty connector is an inbound/outbound factory sharing configuration but usually no * runtime * (connection...) state at the exception of shared connection pool setups. Subscribing * to the returned {@link Mono} will effectively * create a new stateful "client" or "server" socket depending on the implementation. * It might also be working on top of a socket pool or connection pool as well,but the * state should be safely handled by the pool itself. * <p> * <p>Clients or Receivers will onSubscribe when their connection is established. They * will complete when the unique returned closing {@link Publisher} completes itself or if * the connection is remotely terminated. Calling the returned {@link * disposable#dispose()} from {@link Mono#subscribe()} will terminate the subscription * and underlying connection from the local peer. * <p> * <p>Servers or Producers will onSubscribe when their socket is bound locally. They will * never complete as many {@link Publisher} close selectors will be expected. disposing * the returned {@link Mono} will safely call shutdown. * * @param <INBOUND> incoming traffic API such as server request or client response * @param <OUTBOUND> outgoing traffic API such as server response or client request * @author Stephane Maldini * @since 0.6 */ public interface NettyConnector<INBOUND extends NettyInbound,OUTBOUND extends NettyOutbound> { /** * Prepare a {@link BiFunction} IO handler that will react on a new connected state * each * time * the returned {@link Mono} is subscribed. This {@link NettyConnector} shouldn't assume * any state related to the individual created/cleaned resources. * <p> * The IO handler will return {@link Publisher} to signal when to terminate the * underlying resource channel. * * @param ioHandler the in/out callback returning a closing publisher * * @return a {@link Mono} completing with a {@link disposable} token to dispose * the active handler (server,client connection...) or failing with the connection * error. */ Mono<? extends NettyContext> newHandler(BiFunction<? super INBOUND,? super OUTBOUND,? extends Publisher<Void>> ioHandler); /** * Start a Client or Server in a blocking fashion,and wait for it to finish initializing. * The returned {@link BlockingNettyContext} class offers a simplified API around operating * the client/server in a blocking fashion,including to {@link BlockingNettyContext#shutdown() shut it down}. * * @param handler the handler to start the client or server with. * @param <T> * @return a {@link BlockingNettyContext} */ default <T extends BiFunction<INBOUND,OUTBOUND,? extends Publisher<Void>>> BlockingNettyContext start(T handler) { return new BlockingNettyContext(newHandler(handler),getClass().getSimpleName()); } /** * Start a Client or Server in a blocking fashion,including to {@link BlockingNettyContext#shutdown() shut it down}. * * @param handler the handler to start the client or server with. * @param timeout wait for Client/Server to start for the specified timeout. * @param <T> * @return a {@link BlockingNettyContext} */ default <T extends BiFunction<INBOUND,? extends Publisher<Void>>> BlockingNettyContext start(T handler,Duration timeout) { return new BlockingNettyContext(newHandler(handler),getClass().getSimpleName(),timeout); } /** * Start a Client or Server in a fully blocking fashion,not only waiting for it to * initialize but also blocking during the full lifecycle of the client/server. * Since most servers will be long-lived,this is more adapted to running a server * out of a main method,only allowing shutdown of the servers through sigkill. * <p> * Note that a {@link Runtime#addShutdownHook(Thread) JVM shutdown hook} is added * by this method in order to properly disconnect the client/server upon receiving * a sigkill signal. * * @param handler the handler to execute. */ default <T extends BiFunction<INBOUND,? extends Publisher<Void>>> void startAndAwait(T handler) { startAndAwait(handler,null); } /** * Start a Client or Server in a fully blocking fashion,only allowing shutdown of the servers through sigkill. * <p> * Note that a {@link Runtime#addShutdownHook(Thread) JVM shutdown hook} is added * by this method in order to properly disconnect the client/server upon receiving * a sigkill signal. * * @param handler the handler to execute. * @param onStart an optional callback to be invoked once the client/server has finished * initializing. */ default <T extends BiFunction<INBOUND,? extends Publisher<Void>>> void startAndAwait(T handler,@Nullable Consumer<BlockingNettyContext> onStart) { BlockingNettyContext facade = new BlockingNettyContext(newHandler(handler),getClass().getSimpleName()); facade.installShutdownHook(); if (onStart != null) { onStart.accept(facade); } facade.getContext() .onClose() .block(); } }
可以看到这个类有5个方法，一个newHandler是non-blocking模式的，其他的几个start开头的都是blocking模式的(
duration参数用于指定等待初始化完成的超时时间)，使用的是BlockingNettyContext
newHandler
newHandler返回的是一个Mono<? extends NettyContext>，在这个mono完成的时候，会自己dispose。
实例如下
@Test
public void testNewHandler() throws InterruptedException {
TcpClient client = TcpClient.create("localhost",9090);
Mono<? extends NettyContext> mono = client.newHandler((inbound,outbound) -> {
return outbound.sendString(Mono.just("Hello World!")).then();
});
CountDownLatch latch = new CountDownLatch(1);
disposable disposable = mono
.doFinally(e -> {
System.out.println("finish:"+e);
latch.countDown();
})
.subscribe();
latch.await();
System.out.println(disposable.isdisposed());
}
start
start方法返回的是BlockingNettyContext，用户可以调用BlockingNettyContext的shutdown方法来dispose nettyContext，比如
@Test
public void testShutdown(){
TcpClient client = TcpClient.create("localhost",9090);
CountDownLatch latch = new CountDownLatch(1);
BlockingNettyContext context = client.start((inbound,outbound) -> {
latch.countDown();
return outbound.sendString(Mono.just("hello world"))
.then();
});
try {
latch.await();
} catch (InterruptedException e) {
e.printstacktrace();
} finally {
context.shutdown();
}
}
reactor-netty-0.7.6.RELEASE-sources.jar!/reactor/ipc/netty/tcp/BlockingNettyContext.java
/**
* Shut down the {@link NettyContext} and wait for its termination,up to the
* {@link #setLifecycleTimeout(Duration) lifecycle timeout}.
*/
public void shutdown() {
if (context.isdisposed()) {
return;
}
removeShutdownHook(); //only applies if not called from the hook's thread
context.dispose();
context.onClose()
.doOnError(e -> LOG.error("Stopped {} on {} with an error {}",description,context.address(),e))
.doOnTerminate(() -> LOG.info("Stopped {} on {}",context.address()))
.timeout(lifecycleTimeout,Mono.error(new TimeoutException(description + " Couldn't be stopped within " + lifecycleTimeout.toMillis() + "ms")))
.block();
}
/**
* Remove a {@link Runtime#removeShutdownHook(Thread) JVM shutdown hook} if one was
* {@link #installShutdownHook() installed} by this {@link BlockingNettyContext}.
*
* @return true if there was a hook and it was removed,false otherwise.
*/
public boolean removeShutdownHook() {
if (this.shutdownHook != null && Thread.currentThread() != this.shutdownHook) {
Thread sdh = this.shutdownHook;
this.shutdownHook = null;
return Runtime.getRuntime().removeShutdownHook(sdh);
}
return false;
}
这里的shutdown主要是移除当前的shutdownHook,然后dispose nettyContext
startAndAwait
startAndAwait方法调用了BlockingNettyContext的installShutdownHook来进行关闭
reactor-netty-0.7.6.RELEASE-sources.jar!/reactor/ipc/netty/tcp/BlockingNettyContext.java
/**
* Install a {@link Runtime#addShutdownHook(Thread) JVM shutdown hook} that will
* shutdown this {@link BlockingNettyContext} if the JVM is terminated externally.
* <p>
* The hook is removed if shutdown manually,and subsequent calls to this method are
* no-op.
*/
public void installShutdownHook() {
//don't return the hook to discourage uninstalling it externally
if (this.shutdownHook != null) {
return;
}
this.shutdownHook = new Thread(this::shutdownFromJVM);
Runtime.getRuntime().addShutdownHook(this.shutdownHook);
}
protected void shutdownFromJVM() {
if (context.isdisposed()) {
return;
}
final String hookDesc = Thread.currentThread().toString();
context.dispose();
context.onClose()
.doOnError(e -> LOG.error("Stopped {} on {} with an error {} from JVM hook {}",e,hookDesc))
.doOnTerminate(() -> LOG.info("Stopped {} on {} from JVM hook {}",hookDesc))
.timeout(lifecycleTimeout,Mono.error(new TimeoutException(description +
" Couldn't be stopped within " + lifecycleTimeout.toMillis() + "ms")))
.block();
}
在shutdownHook里头注册了shutdownFromJVM方法，用于关闭NettyContext。
实例
@Test
public void testStartAndAwait(){
TcpClient client = TcpClient.create("localhost",9090);
client.startAndAwait((inbound,outbound) -> {
return outbound.sendString(Mono.just("hello world"))
.then();
});
}
小结
NettyConnector提供了non-blocking及blocking两种使用方式，non-blocking的话，使用newHandler返回一个Mono<? extends NettyContext>，在它会在完成的时候，自己dispose nettyContext；blocking的话，startAndAwait方法会自动帮你注册shutdownHook来dispose nettyContext，而start方法则返回BlockingNettyContext，允许调用shutdown方法来dispose nettyContext。
