共计 7047 个字符，预计需要花费 18 分钟才能阅读完成。
序
本文主要研究一下 jdk httpclient 的 executor
HttpClientImpl
java.net.http/jdk/internal/net/http/HttpClientImpl.java
private HttpClientImpl(HttpClientBuilderImpl builder,
SingleFacadeFactory facadeFactory) {
id = CLIENT_IDS.incrementAndGet();
dbgTag = “HttpClientImpl(” + id +”)”;
if (builder.sslContext == null) {
try {
sslContext = SSLContext.getDefault();
} catch (NoSuchAlgorithmException ex) {
throw new InternalError(ex);
}
} else {
sslContext = builder.sslContext;
}
Executor ex = builder.executor;
if (ex == null) {
ex = Executors.newCachedThreadPool(new DefaultThreadFactory(id));
isDefaultExecutor = true;
} else {
isDefaultExecutor = false;
}
delegatingExecutor = new DelegatingExecutor(this::isSelectorThread, ex);
facadeRef = new WeakReference<>(facadeFactory.createFacade(this));
client2 = new Http2ClientImpl(this);
cookieHandler = builder.cookieHandler;
connectTimeout = builder.connectTimeout;
followRedirects = builder.followRedirects == null ?
Redirect.NEVER : builder.followRedirects;
this.userProxySelector = Optional.ofNullable(builder.proxy);
this.proxySelector = userProxySelector
.orElseGet(HttpClientImpl::getDefaultProxySelector);
if (debug.on())
debug.log(“proxySelector is %s (user-supplied=%s)”,
this.proxySelector, userProxySelector.isPresent());
authenticator = builder.authenticator;
if (builder.version == null) {
version = HttpClient.Version.HTTP_2;
} else {
version = builder.version;
}
if (builder.sslParams == null) {
sslParams = getDefaultParams(sslContext);
} else {
sslParams = builder.sslParams;
}
connections = new ConnectionPool(id);
connections.start();
timeouts = new TreeSet<>();
try {
selmgr = new SelectorManager(this);
} catch (IOException e) {
// unlikely
throw new InternalError(e);
}
selmgr.setDaemon(true);
filters = new FilterFactory();
initFilters();
assert facadeRef.get() != null;
}
这里如果 HttpClientBuilderImpl 的 executor 为 null，则会创建 Executors.newCachedThreadPool(new DefaultThreadFactory(id))
HttpClientImpl.sendAsync
java.net.http/jdk/internal/net/http/HttpClientImpl.java
@Override
public <T> CompletableFuture<HttpResponse<T>>
sendAsync(HttpRequest userRequest, BodyHandler<T> responseHandler)
{
return sendAsync(userRequest, responseHandler, null);
}
@Override
public <T> CompletableFuture<HttpResponse<T>>
sendAsync(HttpRequest userRequest,
BodyHandler<T> responseHandler,
PushPromiseHandler<T> pushPromiseHandler) {
return sendAsync(userRequest, responseHandler, pushPromiseHandler, delegatingExecutor.delegate);
}
private <T> CompletableFuture<HttpResponse<T>>
sendAsync(HttpRequest userRequest,
BodyHandler<T> responseHandler,
PushPromiseHandler<T> pushPromiseHandler,
Executor exchangeExecutor) {
Objects.requireNonNull(userRequest);
Objects.requireNonNull(responseHandler);
AccessControlContext acc = null;
if (System.getSecurityManager() != null)
acc = AccessController.getContext();
// Clone the, possibly untrusted, HttpRequest
HttpRequestImpl requestImpl = new HttpRequestImpl(userRequest, proxySelector);
if (requestImpl.method().equals(“CONNECT”))
throw new IllegalArgumentException(“Unsupported method CONNECT”);
long start = DEBUGELAPSED ? System.nanoTime() : 0;
reference();
try {
if (debugelapsed.on())
debugelapsed.log(“ClientImpl (async) send %s”, userRequest);
// When using sendAsync(…) we explicitly pass the
// executor’s delegate as exchange executor to force
// asynchronous scheduling of the exchange.
// When using send(…) we don’t specify any executor
// and default to using the client’s delegating executor
// which only spawns asynchronous tasks if it detects
// that the current thread is the selector manager
// thread. This will cause everything to execute inline
// until we need to schedule some event with the selector.
Executor executor = exchangeExecutor == null
? this.delegatingExecutor : exchangeExecutor;
MultiExchange<T> mex = new MultiExchange<>(userRequest,
requestImpl,
this,
responseHandler,
pushPromiseHandler,
acc);
CompletableFuture<HttpResponse<T>> res =
mex.responseAsync(executor).whenComplete((b,t) -> unreference());
if (DEBUGELAPSED) {
res = res.whenComplete(
(b,t) -> debugCompleted(“ClientImpl (async)”, start, userRequest));
}
// makes sure that any dependent actions happen in the CF default
// executor. This is only needed for sendAsync(…), when
// exchangeExecutor is non-null.
if (exchangeExecutor != null) {
res = res.whenCompleteAsync((r, t) -> {/* do nothing */}, ASYNC_POOL);
}
return res;
} catch(Throwable t) {
unreference();
debugCompleted(“ClientImpl (async)”, start, userRequest);
throw t;
}
}
这里如果是 sendAsync 的话，executor 参数传递的是 delegatingExecutor.delegate；如果是同步的 send 方法，则 executor 传的值是 null
这里创建了一个 MultiExchange，然后调用 mex.responseAsync(executor).whenComplete((b,t) -> unreference())，这里使用了 executor
MultiExchange.responseAsync
java.net.http/jdk/internal/net/http/MultiExchange.java
public CompletableFuture<HttpResponse<T>> responseAsync(Executor executor) {
CompletableFuture<Void> start = new MinimalFuture<>();
CompletableFuture<HttpResponse<T>> cf = responseAsync0(start);
start.completeAsync(() -> null, executor); // trigger execution
return cf;
}
private CompletableFuture<HttpResponse<T>>
responseAsync0(CompletableFuture<Void> start) {
return start.thenCompose(v -> responseAsyncImpl())
.thenCompose((Response r) -> {
Exchange<T> exch = getExchange();
return exch.readBodyAsync(responseHandler)
.thenApply((T body) -> {
this.response =
new HttpResponseImpl<>(r.request(), r, this.response, body, exch);
return this.response;
});
});
}
可以看到这里使用的是 CompletableFuture 的 completeAsync 方法 (注意这个方法是 java9 才有的)，executor 也是在这里使用的
由于默认是使用 Executors.newCachedThreadPool 创建的 executor，要注意控制并发数及任务执行时间，防止线程数无限制增长过度消耗系统资源
/**
* Creates a thread pool that creates new threads as needed, but
* will reuse previously constructed threads when they are
* available, and uses the provided
* ThreadFactory to create new threads when needed.
*
* @param threadFactory the factory to use when creating new threads
* @return the newly created thread pool
* @throws NullPointerException if threadFactory is null
*/
public static ExecutorService newCachedThreadPool(ThreadFactory threadFactory) {
return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
60L, TimeUnit.SECONDS,
new SynchronousQueue<Runnable>(),
threadFactory);
}
RejectedExecutionException
实例代码
@Test
public void testAsyncPool(){
ThreadPoolExecutor executor = ThreadPoolBuilder.fixedPool()
.setPoolSize(2)
.setQueueSize(5)
.setThreadNamePrefix(“test-“)
.build();
List<CompletableFuture<String>> futureList = IntStream.rangeClosed(1,100)
.mapToObj(i -> new CompletableFuture<String>())
.collect(Collectors.toList());
futureList.stream()
.forEach(future -> {
future.completeAsync(() -> {
try {
TimeUnit.SECONDS.sleep(3);
} catch (InterruptedException e1) {
e1.printStackTrace();
}
return “message”;
},executor);
});
CompletableFuture.allOf(futureList
.toArray(new CompletableFuture<?>[futureList.size()]))
.join();
}
这里创建的是 fixedPool，指定 queueSize 为 5
日志输出
java.util.concurrent.RejectedExecutionException: Task java.util.concurrent.CompletableFuture$AsyncSupply@76b10754 rejected from java.util.concurrent.ThreadPoolExecutor@2bea5ab4[Running, pool size = 2, active threads = 2, queued tasks = 5, completed tasks = 0]
at java.base/java.util.concurrent.ThreadPoolExecutor$AbortPolicy.rejectedExecution(ThreadPoolExecutor.java:2055)
at java.base/java.util.concurrent.ThreadPoolExecutor.reject(ThreadPoolExecutor.java:825)
at java.base/java.util.concurrent.ThreadPoolExecutor.execute(ThreadPoolExecutor.java:1355)
at java.base/java.util.concurrent.CompletableFuture.completeAsync(CompletableFuture.java:2591)
可以看到线程池队列大小起到了限制作用
小结
jdk httpclient 的 executor 在进行异步操作的时候使用，默认创建的是使用 Executors.newCachedThreadPool 创建的 executor，其线程池大小是 Integer.MAX_VALUE，因此在使用的时候要注意，最好是改为有界队列，然后再加上线程池的监控。
doc
java.net.http javadoc
