序
本文主要研究一下AsyncHttpClient的IOExceptionFilter
IOExceptionFilter
org/asynchttpclient/filter/IOExceptionFilter.java
代码语言：javascript复制/**
* This filter is invoked when an {@link java.io.IOException} occurs during an http transaction.
*/
public interface IOExceptionFilter {
/**
* An {@link org.asynchttpclient.AsyncHttpClient} will invoke {@link IOExceptionFilter#filter} and will
* use the returned {@link FilterContext} to replay the {@link org.asynchttpclient.Request} or abort the processing.
*
* @param ctx a {@link FilterContext}
* @param <T> the handler result type
* @return {@link FilterContext}. The {@link FilterContext} instance may not the same as the original one.
* @throws FilterException to interrupt the filter processing.
*/
<T> FilterContext<T> filter(FilterContext<T> ctx) throws FilterException;
}
IOExceptionFilter接口定义了filter方法，在捕获到IOException的时候执行
ResumableIOExceptionFilter
org/asynchttpclient/handler/resumable/ResumableIOExceptionFilter.java
代码语言：javascript复制/**
* Simple {@link org.asynchttpclient.filter.IOExceptionFilter} that replay the current {@link org.asynchttpclient.Request} using a {@link ResumableAsyncHandler}
*/
public class ResumableIOExceptionFilter implements IOExceptionFilter {
public <T> FilterContext<T> filter(FilterContext<T> ctx) {
if (ctx.getIOException() != null && ctx.getAsyncHandler() instanceof ResumableAsyncHandler) {
Request request = ResumableAsyncHandler.class.cast(ctx.getAsyncHandler()).adjustRequestRange(ctx.getRequest());
return new FilterContext.FilterContextBuilder<>(ctx).request(request).replayRequest(true).build();
}
return ctx;
}
}
ResumableIOExceptionFilter实现了IOExceptionFilter接口，它在asyncHandler是ResumableAsyncHandler类型的时候执行adjustRequestRange方法获取新的request，然后再通过new FilterContext.FilterContextBuilder<>(ctx).request(request).replayRequest(true).build()返回
ResumableAsyncHandler
org/asynchttpclient/handler/resumable/ResumableAsyncHandler.java
代码语言：javascript复制 /**
* Invoke this API if you want to set the Range header on your {@link Request} based on the last valid bytes
* position.
*
* @param request {@link Request}
* @return a {@link Request} with the Range header properly set.
*/
public Request adjustRequestRange(Request request) {
Long ri = resumableIndex.get(request.getUrl());
if (ri != null) {
byteTransferred.set(ri);
}
// The Resumable
if (resumableListener != null && resumableListener.length() > 0 && byteTransferred.get() != resumableListener.length()) {
byteTransferred.set(resumableListener.length());
}
RequestBuilder builder = new RequestBuilder(request);
if (request.getHeaders().get(RANGE) == null && byteTransferred.get() != 0) {
builder.setHeader(RANGE, "bytes=" byteTransferred.get() "-");
}
return builder.build();
}
adjustRequestRange方法在byteTransferred不等于resumableListener.length()的时候会重置byteTransferred，对于有range的header则重置其bytes参数值
ResponseFiltersInterceptor
org/asynchttpclient/netty/handler/intercept/ResponseFiltersInterceptor.java
代码语言：javascript复制 public boolean exitAfterProcessingFilters(Channel channel,
NettyResponseFuture<?> future,
AsyncHandler<?> handler,
HttpResponseStatus status,
HttpHeaders responseHeaders) {
FilterContext fc = new FilterContext.FilterContextBuilder().asyncHandler(handler).request(future.getCurrentRequest()).responseStatus(status)
.responseHeaders(responseHeaders).build();
for (ResponseFilter asyncFilter : config.getResponseFilters()) {
try {
fc = asyncFilter.filter(fc);
// FIXME Is it worth protecting against this?
assertNotNull("fc", "filterContext");
} catch (FilterException efe) {
requestSender.abort(channel, future, efe);
}
}
// The handler may have been wrapped.
future.setAsyncHandler(fc.getAsyncHandler());
// The request has changed
if (fc.replayRequest()) {
requestSender.replayRequest(future, fc, channel);
return true;
}
return false;
}
ResponseFiltersInterceptor的exitAfterProcessingFilters方法，对于fc.replayRequest()为true则执行requestSender.replayRequest(future, fc, channel)
replayRequest
org/asynchttpclient/netty/request/NettyRequestSender.java
代码语言：javascript复制 public void replayRequest(final NettyResponseFuture<?> future, FilterContext fc, Channel channel) {
Request newRequest = fc.getRequest();
future.setAsyncHandler(fc.getAsyncHandler());
future.setChannelState(ChannelState.NEW);
future.touch();
LOGGER.debug("nnReplaying Request {}n for Future {}n", newRequest, future);
try {
future.getAsyncHandler().onRetry();
} catch (Exception e) {
LOGGER.error("onRetry crashed", e);
abort(channel, future, e);
return;
}
channelManager.drainChannelAndOffer(channel, future);
sendNextRequest(newRequest, future);
}
NettyRequestSender的replayRequest方法会触发future.getAsyncHandler().onRetry()回调，然后执行channelManager.drainChannelAndOffer(channel, future)及sendNextRequest
applyIoExceptionFiltersAndReplayRequest
org/asynchttpclient/netty/request/NettyRequestSender.java
代码语言：javascript复制 public boolean applyIoExceptionFiltersAndReplayRequest(NettyResponseFuture<?> future, IOException e,
Channel channel) {
boolean replayed = false;
@SuppressWarnings({"unchecked", "rawtypes"})
FilterContext<?> fc = new FilterContext.FilterContextBuilder().asyncHandler(future.getAsyncHandler())
.request(future.getCurrentRequest()).ioException(e).build();
for (IOExceptionFilter asyncFilter : config.getIoExceptionFilters()) {
try {
fc = asyncFilter.filter(fc);
assertNotNull(fc, "filterContext");
} catch (FilterException efe) {
abort(channel, future, efe);
}
}
if (fc.replayRequest() && future.incrementRetryAndCheck() && future.isReplayPossible()) {
future.setKeepAlive(false);
replayRequest(future, fc, channel);
replayed = true;
}
return replayed;
}
NettyRequestSender的applyIoExceptionFiltersAndReplayRequest方法会遍历config.getIoExceptionFilters()，然后挨个执行fc = asyncFilter.filter(fc)
小结
AsyncHttpClient的IOExceptionFilter接口定义了filter方法，在捕获到IOException的时候执行；ResumableIOExceptionFilter实现了IOExceptionFilter接口，它在asyncHandler是ResumableAsyncHandler类型的时候执行adjustRequestRange方法获取新的request，然后再通过new FilterContext.FilterContextBuilder<>(ctx).request(request).replayRequest(true).build()返回；NettyRequestSender的replayRequest方法会触发future.getAsyncHandler().onRetry()回调，然后执行channelManager.drainChannelAndOffer(channel, future)及sendNextRequest。
