Reactor-Netty系列1-TcpClient原始碼分析-從示例程式開始
阿新 • • 發佈：2019-12-31
1、示例程式：
Reactor-Netty 版本：
<dependency>
<groupId>io.projectreactor.netty</groupId>
<artifactId>reactor-netty</artifactId>
<version>0.8.10.RELEASE</version>
</dependency>
複製程式碼示例程式：
public class TcpServerApplication {
public static void main(String[] args) TcpServerApplication 輸出結果：
[ INFO] (reactor-tcp-nio-2) onSubscribe(FluxHandle.HandleSubscriber)
[ INFO] (reactor-tcp-nio-2) request(unbounded)
[ INFO] (reactor-tcp-nio-2) onNext(Hello World!)
[ INFO] (reactor-tcp-nio-2) cancel()
複製程式碼基本邏輯是：Server 端繫結 8080 埠並監聽請求；Client 端連線上埠後傳送字串 Hello World!；Server 埠收到請求後打印出來。
下面進行具體原始碼分析。
2、TcpClient
TcpClient.create()
public static TcpClient create() {
return create(TcpResources.get());
}
/**
* 最終返回的是 TcpClientConnect
* 從入參可知，TcpClientConnect 關注的是連線管理 ConnectionProvider
*/
public static TcpClient create(ConnectionProvider provider) {
return new TcpClientConnect(provider);
}
public class TcpResources implements ConnectionProvider,LoopResources {
final ConnectionProvider defaultProvider;
final LoopResources defaultLoops;
protected TcpResources(LoopResources defaultLoops,ConnectionProvider defaultProvider) {
this.defaultLoops = defaultLoops;
this.defaultProvider = defaultProvider;
}
/**
* 該靜態方法最終返回的是 TcpResources，包括:
* ConnectionProvider: 管理連線
* LoopResources: 管理執行緒
*/
public static TcpResources get() {
// 如果不存在，那麼建立 TcpResources；否則，直接返回 TcpResources
return getOrCreate(tcpResources,null,ON_TCP_NEW,"tcp");
}
複製程式碼host()
/**
* 1. 最終返回的是 TcpClientBootstrap
* 2. TcpClientBootstrap 類有一個 bootstrapMapper,是一個 Function： b -> TcpUtils.updateHost(b,host)，關注兩個地方：b 是一個 Bootstrap 物件，b 何時生成？Function 介面的 apply 方法什麼時候被執行？可以看到 TcpClientBootstrap 類的 configure() 方法同時滿足了上面 2 個地方，因此只需要關注該方法何時被呼叫即可。
*/
public final TcpClient host(String host) {
Objects.requireNonNull(host,"host");
return bootstrap(b -> TcpUtils.updateHost(b,host));
}
public final TcpClient bootstrap(Function<? super Bootstrap,? extends Bootstrap> bootstrapMapper) {
return new TcpClientBootstrap(this,bootstrapMapper);
}
final class TcpClientBootstrap extends TcpClientOperator {
final Function<? super Bootstrap,? extends Bootstrap> bootstrapMapper;
TcpClientBootstrap(TcpClient client,Function<? super Bootstrap,? extends Bootstrap> bootstrapMapper) {
super(client);
this.bootstrapMapper = Objects.requireNonNull(bootstrapMapper,"bootstrapMapper");
}
@Override
public Bootstrap configure() {
return Objects.requireNonNull(bootstrapMapper.apply(source.configure()),"bootstrapMapper");
}
}
複製程式碼port()
/**
* 和 host(String host) 方法類似
*/
public final TcpClient port(int port) {
return bootstrap(b -> TcpUtils.updatePort(b,port));
}
複製程式碼handler()
/**
* 最終返回的是 TcpClientDoOn；
* handler 的入參是 BiFunction，並且在 doOnConnected 方法中直接呼叫了 apply 方法；
* BiFunction 返回的 Publisher 也直接呼叫了 subscribe 方法；
* 因此，只需要關注 doOnConnected 方法的入參 Consumer 何時被呼叫即可
*/
public final TcpClient handle(BiFunction<? super NettyInbound,? super NettyOutbound,? extends Publisher<Void>> handler) {
Objects.requireNonNull(handler,"handler");
return doOnConnected(c -> {
if (log.isDebugEnabled()) {
log.debug(format(c.channel(),"Handler is being applied: {}"),handler);
}
Mono.fromDirect(handler.apply((NettyInbound) c,(NettyOutbound) c))
.subscribe(c.disposeSubscriber());
});
}
public final TcpClient doOnConnected(Consumer<? super Connection> doOnConnected) {
Objects.requireNonNull(doOnConnected,"doOnConnected");
return new TcpClientDoOn(this,doOnConnected,null);
}
final class TcpClientDoOn extends TcpClientOperator implements ConnectionObserver {
final Consumer<? super Bootstrap> onConnect;
// onConnected 即 handle 方法中呼叫的 doOnConnected 的 Consumer
final Consumer<? super Connection> onConnected;
final Consumer<? super Connection> onDisconnected;
TcpClientDoOn(TcpClient client,@Nullable Consumer<? super Bootstrap> onConnect,@Nullable Consumer<? super Connection> onConnected,@Nullable Consumer<? super Connection> onDisconnected) {
// 繼承上一個 TcpClient
super(client);
this.onConnect = onConnect;
this.onConnected = onConnected;
this.onDisconnected = onDisconnected;
}
@Override
public Bootstrap configure() {
Bootstrap b = source.configure();
ConnectionObserver observer = BootstrapHandlers.connectionObserver(b);
// 注意：這裡設定了 ConnectionObserver，後面會講到
BootstrapHandlers.connectionObserver(b,observer.then(this));
return b;
}
@Override
public Mono<? extends Connection> connect(Bootstrap b) {
if (onConnect != null) {
return source.connect(b)
.doOnSubscribe(s -> onConnect.accept(b));
}
return source.connect(b);
}
@Override
public void onStateChange(Connection connection,State newState) {
// onConnected 在這裡被呼叫，即 connection 狀態改變時
if (onConnected != null && newState == State.CONFIGURED) {
onConnected.accept(connection);
return;
}
if (onDisconnected != null) {
if (newState == State.DISCONNECTING) {
connection.onDispose(() -> onDisconnected.accept(connection));
}
else if (newState == State.RELEASED) {
onDisconnected.accept(connection);
}
}
}
}
複製程式碼connectNow()
// 設定超時 45s
public final Connection connectNow() {
return connectNow(Duration.ofSeconds(45));
}
public final Connection connectNow(Duration timeout) {
Objects.requireNonNull(timeout,"timeout");
try {
// 這裡 connect() 方法返回的是 Mono
return Objects.requireNonNull(connect().block(timeout),"aborted");
}
catch (IllegalStateException e) {
...
}
}
// 返回的是 Mono
public final Mono<? extends Connection> connect() {
...
return connect(b);
}
// block 方法中直接開始訂閱
public T block(Duration timeout) {
BlockingMonoSubscriber<T> subscriber = new BlockingMonoSubscriber<>();
onLastAssembly(this).subscribe(Operators.toCoreSubscriber(subscriber));
return subscriber.blockingGet(timeout.toMillis(),TimeUnit.MILLISECONDS);
}
final T blockingGet(long timeout,TimeUnit unit) {
...
if (getCount() != 0) {
try {
if (!await(timeout,unit)) {
dispose(); // 超時取消訂閱
throw new IllegalStateException("Timeout on blocking read for " + timeout + " " + unit);
}
}
catch (InterruptedException ex) {
dispose();
RuntimeException re = Exceptions.propagate(ex);
//this is ok,as re is always a new non-singleton instance
re.addSuppressed(new Exception("#block has been interrupted"));
throw re;
}
}
...
}
複製程式碼由以上分析可知，在最後的 connectNow() 方法中，才開始真正的訂閱執行。下面繼續分析 connect 方法。
connect()
public final Mono<? extends Connection> connect() {
Bootstrap b;
try {
// 1. 獲取預設的 Bootstrap
b = configure();
}
catch (Throwable t) {
Exceptions.throwIfJvmFatal(t);
return Mono.error(t);
}
// 2. connect(b)
return connect(b);
}
public Bootstrap configure() {
return DEFAULT_BOOTSTRAP.clone();
}
static final Bootstrap DEFAULT_BOOTSTRAP =
new Bootstrap().option(ChannelOption.AUTO_READ,false) .remoteAddress(InetSocketAddressUtil.createUnresolved(NetUtil.LOCALHOST.getHostAddress(),DEFAULT_PORT));
複製程式碼繼續看 connect(Bootstrap b) 方法：
// 這是一個抽象方法，很多繼承類都實現了該方法。根據之前的程式碼分析，首先呼叫的應該是 TcpClientDoOn 類
public abstract Mono<? extends Connection> connect(Bootstrap b);
// TcpClientDoOn 類
public Mono<? extends Connection> connect(Bootstrap b) {
if (onConnect != null) {
return source.connect(b)
.doOnSubscribe(s -> onConnect.accept(b));
}
// 往上傳遞，source 代表上一個 TcpClient；最終傳遞到初始的 TcpClientConnect
return source.connect(b);
}
// TcpClientConnect 類
final ConnectionProvider provider;
public Mono<? extends Connection> connect(Bootstrap b) {
// 填充 b 的屬性
if (b.config()
.group() == null) {
TcpClientRunOn.configure(b,LoopResources.DEFAULT_NATIVE,TcpResources.get(),maxConnections != -1);
}
// 最終呼叫這個方法
return provider.acquire(b);
}
複製程式碼ConnectionProvider
上面講到 connect 方法最終呼叫的是 ConnectionProvider 類中的方法。ConnectionProvider 在之前的分析中出現過，即TcpResources.get() 方法返回的 TcpResources 物件中包含這個屬性。
// 建立預設的 TcpResources
static <T extends TcpResources> T create(@Nullable T previous,@Nullable LoopResources loops,@Nullable ConnectionProvider provider,String name,BiFunction<LoopResources,ConnectionProvider,T> onNew) {
if (previous == null) {
loops = loops == null ? LoopResources.create("reactor-" + name) : loops;
// 建立 ConnectionProvider
provider = provider == null ? ConnectionProvider.elastic(name) : provider;
}
else {
loops = loops == null ? previous.defaultLoops : loops;
provider = provider == null ? previous.defaultProvider : provider;
}
return onNew.apply(loops,provider);
}
}
static ConnectionProvider elastic(String name) {
// 這裡的第 2 個入參 PoolFactory 又是一個函式式介面，因此物件的生成時間點在於何時呼叫 PoolFactory.newPool 方法; 生成的 ChannelPool 型別為 SimpleChannelPool。
return new PooledConnectionProvider(name,(bootstrap,handler,checker) -> new SimpleChannelPool(bootstrap,checker,true,false));
}
final class PooledConnectionProvider implements ConnectionProvider {
interface PoolFactory {
ChannelPool newPool(Bootstrap b,ChannelPoolHandler handler,ChannelHealthChecker checker);
}
final ConcurrentMap<PoolKey,Pool> channelPools;
final String name;
final PoolFactory poolFactory;
final int maxConnections;
PooledConnectionProvider(String name,PoolFactory poolFactory) {
this.name = name;
this.poolFactory = poolFactory;
this.channelPools = PlatformDependent.newConcurrentHashMap();
this.maxConnections = -1;
}
...
}
複製程式碼現在回到 provider.acquire(b) 方法，可以知道呼叫的是 PooledConnectionProvider 類中的方法，繼續分析：
// Map 結構，每個 (remote address,handler) 組合都有一個連線池
final ConcurrentMap<PoolKey,Pool> channelPools;
final String name;
// 通過 poolFactory 生成 ChannelPool
final PoolFactory poolFactory;
final int maxConnections;
/**
* 主要作用是從連線池中獲取連線
* 首先需要找到對應的連線池,通過 channelPools.get(holder)
* 如果不存在，那麼建立新的連線池，並加入到 channelPools 中
* 最後呼叫 disposableAcquire(sink,obs,pool,false);
*/
public Mono<Connection> acquire(Bootstrap b) {
return Mono.create(sink -> {
Bootstrap bootstrap = b.clone();
// TODO:
ChannelOperations.OnSetup opsFactory =
BootstrapHandlers.channelOperationFactory(bootstrap);
// TODO: 連線生命週期的監聽器
ConnectionObserver obs = BootstrapHandlers.connectionObserver(bootstrap);
// 懶載入，這裡需要設定 bootstrap 的 remote address(ip:port)
NewConnectionProvider.convertLazyRemoteAddress(bootstrap);
// 每個 (remote address,handler) 都有一個 Pool
ChannelHandler handler = bootstrap.config().handler();
PoolKey holder = new PoolKey(bootstrap.config().remoteAddress(),handler != null ? handler.hashCode() : -1);
Pool pool;
for (; ; ) {
// 直接獲取
pool = channelPools.get(holder);
if (pool != null) {
break;
}
// 不存在則建立新的連線池
pool = new Pool(bootstrap,poolFactory,opsFactory);
if (channelPools.putIfAbsent(holder,pool) == null) {
if (log.isDebugEnabled()) {
log.debug("Creating new client pool [{}] for {}",name,bootstrap.config()
.remoteAddress());
}
break;
}
// 關閉多建立的 pool
pool.close();
}
disposableAcquire(sink,false);
});
}
Pool(Bootstrap bootstrap,PoolFactory provider,ChannelOperations.OnSetup opsFactory) {
this.bootstrap = bootstrap;
this.opsFactory = opsFactory;
// 建立新的連線池
this.pool = provider.newPool(bootstrap,this,this);
this.defaultGroup = bootstrap.config()
.group();
HEALTHY = defaultGroup.next()
.newSucceededFuture(true);
UNHEALTHY = defaultGroup.next()
.newSucceededFuture(false);
}
複製程式碼繼續 disposableAcquire 方法，
static void disposableAcquire(MonoSink<Connection> sink,ConnectionObserver obs,Pool pool,boolean retried) {
// 獲取 Channel
Future<Channel> f = pool.acquire();
DisposableAcquire disposableAcquire =
new DisposableAcquire(sink,f,retried);
// 設定監聽器,該方法最終會呼叫 disposableAcquire.operationComplete() 方法，operationComplete() 方法會呼叫 disposableAcquire.run()
f.addListener(disposableAcquire);
sink.onCancel(disposableAcquire);
}
final static class DisposableAcquire
implements Disposable,GenericFutureListener<Future<Channel>>,ConnectionObserver,Runnable {
final Future<Channel> f;
final MonoSink<Connection> sink;
final Pool pool;
final ConnectionObserver obs;
final boolean retried;
DisposableAcquire(MonoSink<Connection> sink,Future<Channel> future,boolean retried) {
this.f = future;
this.pool = pool;
this.sink = sink;
this.obs = obs;
this.retried = retried;
}
// 當連線的狀態改變時，呼叫 obs.onStateChange；而這裡的 obs 就是我們在 TcpClientDoOn.configure() 方法中設定的；所以一旦連線狀態改變，就會呼叫 TcpClient.handle 中的方法
@Override
public void onStateChange(Connection connection,State newState) {
if (newState == State.CONFIGURED) {
sink.success(connection);
}
obs.onStateChange(connection,newState);
}
...
}
複製程式碼DisposableAcquire 是一個監聽器，監聽的是連線，即上面程式碼中的 Future f = pool.acquire()。那麼這個 f 是什麼型別呢？之前的程式碼分析中已經知道 pool 為 SimpleChannelPool 型別。
public SimpleChannelPool(Bootstrap bootstrap,final ChannelPoolHandler handler,ChannelHealthChecker healthCheck,boolean releaseHealthCheck,boolean lastRecentUsed) {
this.handler = checkNotNull(handler,"handler");
this.healthCheck = checkNotNull(healthCheck,"healthCheck");
this.releaseHealthCheck = releaseHealthCheck;
// Clone the original Bootstrap as we want to set our own handler
this.bootstrap = checkNotNull(bootstrap,"bootstrap").clone();
this.bootstrap.handler(new ChannelInitializer<Channel>() {
@Override
protected void initChannel(Channel ch) throws Exception {
assert ch.eventLoop().inEventLoop();
// 當新建連線時，會呼叫該方法
handler.channelCreated(ch);
}
});
this.lastRecentUsed = lastRecentUsed;
}
}
public void channelCreated(Channel ch) {
inactiveConnections.incrementAndGet();
...
// 這裡把 ch 包裝成了一下,PooledConnection 這個類同時實現了 Connection 以及 ConnectionObserver 介面，也就是說既是一個 channel，又是一個 listener。後續如果 channel 的狀態發生改變，會呼叫 PooledConnection 的 onStateChange 方法。
PooledConnection pooledConnection = new PooledConnection(ch,this);
pooledConnection.bind();
Bootstrap bootstrap = this.bootstrap.clone();
BootstrapHandlers.finalizeHandler(bootstrap,opsFactory,pooledConnection);
ch.pipeline()
.addFirst(bootstrap.config()
.handler());
}
複製程式碼下面繼續看 PooledConnection 的 onStateChange 方法。
public void onStateChange(Connection connection,State newState) {
if (newState == State.DISCONNECTING) {
...
}
// 其他狀態走這裡
owner().onStateChange(connection,newState);
}
ConnectionObserver owner() {
ConnectionObserver obs;
for (;;) {
obs = channel.attr(OWNER)
.get();
if (obs == null) {
obs = new PendingConnectionObserver();
}
else {
return obs;
}
// 設定 channel.attr(OWNER) 為新建立的 PendingConnectionObserver
// 之後再次呼叫 own() 方法時直接返回該 PendingConnectionObserver
if (channel.attr(OWNER)
.compareAndSet(null,obs)) {
return obs;
}
}
}
final static class PendingConnectionObserver implements ConnectionObserver {
final Queue<Pending> pendingQueue = Queues.<Pending>unbounded(4).get();
@Override
public void onUncaughtException(Connection connection,Throwable error) {
pendingQueue.add(new Pending(connection,error,null));
}
@Override
public void onStateChange(Connection connection,State newState) {
// 把狀態變更放入了等待佇列，其他什麼都不做
pendingQueue.add(new Pending(connection,newState));
}
static class Pending {
final Connection connection;
final Throwable error;
final State state;
Pending(Connection connection,@Nullable Throwable error,@Nullable State state) {
this.connection = connection;
this.error = error;
this.state = state;
}
}
}
複製程式碼從上面程式碼可知，Channel 的狀態變更最終放入了一個等待佇列，缺少了通知各個監聽器的呼叫。繼續回到 DisposableAcquire 類，發現同時實現了 Runnable 介面。
final static class DisposableAcquire
implements Disposable,Runnable {
final Future<Channel> f;
final MonoSink<Connection> sink;
final Pool pool;
final ConnectionObserver obs;
final boolean retried;
@Override
public void onStateChange(Connection connection,newState);
}
@Override
public void run() {
Channel c = f.getNow();
pool.activeConnections.incrementAndGet();
pool.inactiveConnections.decrementAndGet();
// 之前 owner() 方法設定了 PendingConnectionObserver
ConnectionObserver current = c.attr(OWNER)
.getAndSet(this);
if (current instanceof PendingConnectionObserver) {
PendingConnectionObserver pending = (PendingConnectionObserver)current;
PendingConnectionObserver.Pending p;
current = null;
// 監聽連線關閉
registerClose(c,pool);
// 依次處理等待佇列中的事件(連線狀態變更)
while((p = pending.pendingQueue.poll()) != null) {
if (p.error != null) {
onUncaughtException(p.connection,p.error);
}
else if (p.state != null) {
// 通知各個監聽器
onStateChange(p.connection,p.state);
}
}
}
else if (current == null) {
registerClose(c,pool);
}
// TODO: 什麼情況會走這邊？
if (current != null) {
Connection conn = Connection.from(c);
if (log.isDebugEnabled()) {
log.debug(format(c,"Channel acquired,now {} active connections and {} inactive connections"),pool.activeConnections,pool.inactiveConnections);
}
obs.onStateChange(conn,State.ACQUIRED);
PooledConnection con = conn.as(PooledConnection.class);
if (con != null) {
ChannelOperations<?,?> ops = pool.opsFactory.create(con,con,null);
if (ops != null) {
ops.bind();
obs.onStateChange(ops,State.CONFIGURED);
sink.success(ops);
}
else {
//already configured,just forward the connection
sink.success(con);
}
}
else {
//already bound,just forward the connection
sink.success(conn);
}
return;
}
//Connected,leave onStateChange forward the event if factory
...
if (pool.opsFactory == ChannelOperations.OnSetup.empty()) {
sink.success(Connection.from(c));
}
}
}
複製程式碼至此，TcpClient 示例程式中的幾行程式碼差不多就算是分析完了。