reactor-netty中HttpClient对TcpClient的封装-白红宇

序

本文主要研究一下reactor-netty中HttpClient对TcpClien的封装

maven


    			
     
      io.projectreactor.ipc
     			
     
      reactor-netty
     			
     
      0.7.3.RELEASE
     		
    复制代码

实例

HttpClient client = HttpClient.create();        Mono
    
      mono = client.get("http://baidu.com");        //NOTE reactor.ipc.netty.http.client.MonoHttpClientResponse        LOGGER.info("mono resp:{}",mono.getClass());        mono.subscribe();复制代码

HttpClient.request

reactor-netty-0.7.3.RELEASE-sources.jar!/reactor/ipc/netty/http/client/HttpClient.java

/**	 * Use the passed HTTP method to send to the given URL. When connection has been made,	 * the passed handler is invoked and can be used to tune the request and	 * write data to it.	 *	 * @param method the HTTP method to send	 * @param url the target remote URL	 * @param handler the {@link Function} to invoke on opened TCP connection	 * @return a {@link Mono} of the {@link HttpServerResponse} ready to consume for	 * response	 */	public Mono
    
      request(HttpMethod method,			String url,			Function
     > handler) {		if (method == null || url == null) {			throw new IllegalArgumentException("Method && url cannot be both null");		}		return new MonoHttpClientResponse(this, url, method, handler(handler, options));	}复制代码

Mono.subscribe

reactor-core-3.1.3.RELEASE-sources.jar!/reactor/core/publisher/Mono.java

/**	 * Subscribe to this {@link Mono} and request unbounded demand.	 * 	 * This version doesn't specify any consumption behavior for the events from the	 * chain, especially no error handling, so other variants should usually be preferred.	 *	 * 
	 * 	 * 
	 *	 * @return a new {@link Disposable} that can be used to cancel the underlying {@link Subscription}	 */	public final Disposable subscribe() {		if(this instanceof MonoProcessor){			MonoProcessor
     
       s = (MonoProcessor
      
       )this;			s.connect();			return s;		}		else{			return subscribeWith(new LambdaMonoSubscriber<>(null, null, null, null));		}	}复制代码

这里调用了subscribeWith，创建了一个LambdaMonoSubscriber

/**	 * Subscribe the given {@link Subscriber} to this {@link Mono} and return said	 * {@link Subscriber} (eg. a {@link MonoProcessor}).	 *	 * @param subscriber the {@link Subscriber} to subscribe with	 * @param 
    
      the reified type of the {@link Subscriber} for chaining	 *	 * @return the passed {@link Subscriber} after subscribing it to this {@link Mono}	 */	public final 
     
      > E subscribeWith(E subscriber) {		subscribe(subscriber);		return subscriber;	}	public final void subscribe(Subscriber
       actual) {		onLastAssembly(this).subscribe(Operators.toCoreSubscriber(actual));	}	复制代码

这个onLastAssembly(this).subscribe调用的是子类的方法

MonoHttpClientResponse.subscribe

reactor-netty-0.7.3.RELEASE-sources.jar!/reactor/ipc/netty/http/client/MonoHttpClientResponse.java

public void subscribe(final CoreSubscriber
     subscriber) {		ReconnectableBridge bridge = new ReconnectableBridge();		bridge.activeURI = startURI;		Mono.defer(() -> parent.client.newHandler(new HttpClientHandler(this, bridge),				parent.options.getRemoteAddress(bridge.activeURI),				HttpClientOptions.isSecure(bridge.activeURI),				bridge))		    .retry(bridge)		    .cast(HttpClientResponse.class)		    .subscribe(subscriber);	}复制代码

这里使用Mono.defer又对client.newHandler包装了下，defer的英文原意是Defers the creation of the actual Publisher the Subscriber will be subscribed to.，也就是延迟publisher的创建

这里的subscriber便是Operators.toCoreSubscriber(lambdaMonoSubscriber)

可以看到这里调用了parent的client.newHandler，这里的parent便是HttpClient，里头的client是TcpClient

retry使用的是ReconnectableBridge，handler使用的是HttpClientHandler

MonoHttpClientResponse#ReconnectableBridge

static final class ReconnectableBridge			implements Predicate
    
     , Consumer
     
       {		volatile URI      activeURI;		volatile String[] redirectedFrom;		ReconnectableBridge() {		}		void redirect(String to) {			String[] redirectedFrom = this.redirectedFrom;			URI from = activeURI;			try {				activeURI = new URI(to);			}			catch (URISyntaxException e) {				throw Exceptions.propagate(e);			}			if (redirectedFrom == null) {				this.redirectedFrom = new String[]{from.toString()};			}			else {				String[] newRedirectedFrom = new String[redirectedFrom.length + 1];				System.arraycopy(redirectedFrom,						0,						newRedirectedFrom,						0,						redirectedFrom.length);				newRedirectedFrom[redirectedFrom.length] = from.toString();				this.redirectedFrom = newRedirectedFrom;			}		}		@Override		public void accept(Channel channel) {			String[] redirectedFrom = this.redirectedFrom;			if (redirectedFrom != null) {				channel.attr(HttpClientOperations.REDIRECT_ATTR_KEY)				       .set(redirectedFrom);			}		}		@Override		public boolean test(Throwable throwable) {			if (throwable instanceof RedirectClientException) {				RedirectClientException re = (RedirectClientException) throwable;				redirect(re.location);				return true;			}			if (AbortedException.isConnectionReset(throwable)) {				redirect(activeURI.toString());				return true;			}			return false;		}	}复制代码

这里看好像是处理redirect的，并不是真正意义上的retry，比如retry多少次之类的

MonoHttpClientResponse#HttpClientHandler

reactor-netty-0.7.3.RELEASE-sources.jar!/reactor/ipc/netty/http/client/MonoHttpClientResponse.java

static final class HttpClientHandler			implements BiFunction
    
     > {		final MonoHttpClientResponse parent;		final ReconnectableBridge    bridge;		HttpClientHandler(MonoHttpClientResponse parent, ReconnectableBridge bridge) {			this.bridge = bridge;			this.parent = parent;		}		@Override		public Publisher
     
       apply(NettyInbound in, NettyOutbound out) {			try {				URI uri = bridge.activeURI;				HttpClientOperations ch = (HttpClientOperations) in;				String host = uri.getHost();				int port = uri.getPort();				if (port != -1 && port != 80 && port != 443) {					host = host + ':' + port;				}				ch.getNettyRequest()				  .setUri(uri.getRawPath() + (uri.getQuery() == null ? "" :						  "?" + uri.getRawQuery()))				  .setMethod(parent.method)				  .setProtocolVersion(HttpVersion.HTTP_1_1)				  .headers()				  .add(HttpHeaderNames.HOST, host)				  .add(HttpHeaderNames.ACCEPT, ALL);				if (parent.method == HttpMethod.GET						|| parent.method == HttpMethod.HEAD						|| parent.method == HttpMethod.DELETE) {					ch.chunkedTransfer(false);				}				if (parent.handler != null) {					return parent.handler.apply(ch);				}				else {					return ch.send();				}			}			catch (Throwable t) {				return Mono.error(t);			}		}		@Override		public String toString() {			return "HttpClientHandler{" + "startURI=" + bridge.activeURI + ", method=" + parent.method + ", handler=" + parent.handler + '}';		}	}复制代码

这里的handler可以看到netty的痕迹，最后是直接调用HttpClientOperations.send方法

reactor-netty-0.7.3.RELEASE-sources.jar!/reactor/ipc/netty/http/client/HttpClientOperations.java

public Mono
    
      send() {		if (markSentHeaderAndBody()) {			HttpMessage request = newFullEmptyBodyMessage();			return FutureMono.deferFuture(() -> channel().writeAndFlush(request));		}		else {			return Mono.empty();		}	}复制代码

最后调用netty的channel().writeAndFlush(request)将请求发送出去

小结

reactor-netty中的HttpClient对TcpClient进行了桥接，而TcpClient则是基于netty来实现。